tests/test_conversational_needs.py
Rama main · réplica de lectura del repositorio original.
"""Conversational context must reach retrieval without rewriting captured user turns."""
import json
import sys
import tempfile
import threading
import time
import unittest
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from unittest import mock
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
import xkb_agent_hook as hook
import xkb_memory_service as service
import xkb_recall as recall
import xkb_review as review
import distill_memory_to_wiki as distill
import xkb_jev as jev
import continuity_recall as cr
class ConversationalNeeds(unittest.TestCase):
def setUp(self):
temp = tempfile.TemporaryDirectory()
self.addCleanup(temp.cleanup)
self.root = Path(temp.name)
def test_failed_preparation_config_preserves_keyword_candidates(self):
store = service.Store(self.root/'test.sqlite')
wiki = self.root/'topics'
wiki.mkdir()
(wiki/'deployment.md').write_text('# Deployment\n\nDeployment keeps a verified rollback artifact.', encoding='utf-8')
store.catalog.wiki_topics_dir = wiki
with mock.patch.object(jev, 'available', side_effect=FileNotFoundError('missing runtime config')), \
mock.patch.object(jev, 'relevance', side_effect=FileNotFoundError('missing runtime config')):
packet = store.knowledge_recall('deployment', options={'semantic': False, 'cards': False, 'conversations': False})
self.assertTrue(packet['records'])
self.assertEqual(packet['quality']['status'], 'degraded')
self.assertEqual(packet['delivery']['records'], [])
def test_hook_sends_prior_dialogue_for_a_statement_without_changing_capture(self):
transcript = self.root / "conversation.jsonl"
prior = [{"role": "user", "content": "I am building a Seedance character sequence."},
{"role": "assistant", "content": "Keep the character reference stable across shots."}]
current = "The face still changes on the next shot."
transcript.write_text("\n".join(json.dumps({"message": m}) for m in
prior + [{"role": "user", "content": current}]), encoding="utf-8")
with mock.patch.object(hook, "STATE_DIR", self.root / "state"), \
mock.patch.object(hook, "call", side_effect=[{"session_id": "s"}, {"retrieval": {"records": []}}]) as call:
hook.on_prompt({"prompt": current, "session_id": "s", "transcript_path": str(transcript)}, {"source": "test"})
captured = json.loads(hook.state_path("s").read_text(encoding="utf-8"))
body = call.call_args.args[1]
self.assertEqual(body["query"], current)
self.assertEqual(body["conversation"], prior)
self.assertEqual(captured["query"], current)
def test_context_is_bounded_and_tools_and_harness_are_not_conversation(self):
recent = recall.conversation_messages([{"role": "user", "content": "x" * 2000}] * 8)
self.assertEqual(len(recent), 4)
self.assertTrue(all(len(m["content"]) == 600 for m in recent))
with self.assertRaises(ValueError):
recall.conversation_messages([{"role": "tool", "content": "tool output"}])
def test_hook_allows_recall_to_finish_after_metadata_request_budget(self):
class Handler(BaseHTTPRequestHandler):
def do_POST(self):
time.sleep(.1)
self.send_response(200)
self.end_headers()
self.wfile.write(b'{"retrieval":{"records":[]}}')
def log_message(self, *args):
pass
server = ThreadingHTTPServer(('127.0.0.1', 0), Handler)
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
try:
with mock.patch.object(hook, 'TIMEOUT_SECONDS', .01), \
mock.patch.object(hook, 'RECALL_TIMEOUT_SECONDS', 1):
result = hook.call('/v1/turns/start', {'query': 'It still changes between shots.'},
{'url': f'http://127.0.0.1:{server.server_port}', 'token': ''})
self.assertEqual(result['retrieval']['records'], [])
finally:
server.shutdown()
server.server_close()
thread.join()
def test_core_uses_context_for_statements_but_suppresses_acknowledgement(self):
store = service.Store(self.root / "knowledge.sqlite")
history = [{"role": "user", "content": "Seedance character reference"}]
with mock.patch.object(store.catalog, "search", return_value={"records": []}) as search, \
mock.patch.object(store, "recall", return_value={"memories": []}), \
mock.patch.dict("os.environ", {"XKB_JEV_DECIDE": "0"}):
packet = store.knowledge_recall("It still changes between shots.", conversation=history)
self.assertIn("Seedance", search.call_args.args[0])
self.assertEqual(packet["query"], "It still changes between shots.")
self.assertEqual(packet["conversation_messages"], 1)
search.reset_mock()
packet = store.knowledge_recall("ok", conversation=history)
self.assertEqual(packet["retrieval_mode"], "skipped")
search.assert_not_called()
def test_session_context_never_crosses_session_boundary(self):
store = service.Store(self.root / "knowledge.sqlite")
first = store.open_session({"source": "test", "session_key": "one"})["session_id"]
other = store.open_session({"source": "test", "session_key": "two"})["session_id"]
with mock.patch.object(store, "knowledge_recall", return_value={"records": []}) as run:
store.start_turn({"session_id": first, "turn_id": "t1", "query": "Seedance character"})
store.complete_turn("t1", {"session_id": first, "query": "Seedance character", "answer": "Use character references."})
store.start_turn({"session_id": first, "turn_id": "t2", "query": "It still changes."})
self.assertEqual(run.call_args.kwargs["conversation"][0]["content"], "Seedance character")
store.start_turn({"session_id": other, "turn_id": "t3", "query": "It still changes."})
self.assertEqual(run.call_args.kwargs["conversation"], [])
def test_current_need_and_followup_context_both_reach_judge(self):
store = service.Store(self.root / "knowledge.sqlite")
def search(query, *args, **kwargs):
key = "old-video" if "Seedance" in query else "carbon"
return {"records": [{"id": key, "record_type": "knowledge_card", "summary": key,
"namespace": "private", "score": 0.8}]}
def judge(query, records):
self.assertEqual({r['id'] for r in records}, {'carbon', 'old-video'})
return [r for r in records if r['id'] == 'carbon'], {"status": "judged"}
with mock.patch.object(store.catalog, "search", side_effect=search), \
mock.patch.object(store, "recall", return_value={"memories": []}), \
mock.patch.object(service, "judge_relevance", side_effect=judge), \
mock.patch.dict("os.environ", {"XKB_JEV_DECIDE": "1"}):
result = store.knowledge_recall("carbon emissions", 1,
conversation=[{"role": "user", "content": "Seedance camera shots"}])
self.assertEqual([r['id'] for r in result['records']], ['carbon'])
def test_topic_shift_retrieves_new_need_without_losing_prior_references(self):
store = service.Store(self.root / "knowledge.sqlite")
current = "影片先這樣。另外待審草稿越積越多，我沒時間全看"
history = [{"role": "user", "content": "Seedance character sequence"}]
barrier = threading.Barrier(3)
seen = []
def search(query, limit, namespace, **kwargs):
seen.append(query)
barrier.wait(timeout=5) # All bounded branches can run independently.
key = "queue" if query.startswith("另外") else "video"
return {"records": [{"id": key, "summary": key, "score": .8,
"record_type": "knowledge_card", "namespace": namespace}],
"backends": {"wiki": {"status": "error" if key == "queue" else "used"}}}
def judge(query, records):
self.assertIn(current, query)
self.assertIn(history[0]['content'], query)
self.assertEqual({r['id'] for r in records}, {'queue', 'video'})
return [], {"status": "judged"}
with mock.patch.object(store.catalog, "search", side_effect=search), \
mock.patch.object(store, "recall", return_value={"memories": []}), \
mock.patch.object(service, "judge_relevance", side_effect=judge), \
mock.patch.dict("os.environ", {"XKB_JEV_DECIDE": "1"}):
packet = store.knowledge_recall(current, conversation=history)
self.assertEqual(len(seen), 3)
self.assertIn(current, seen)
self.assertTrue(any("Seedance" in q for q in seen))
self.assertEqual(packet['query'], current)
self.assertEqual(len(packet['retrieval_branches']), 3)
self.assertEqual(packet['quality']['status'], 'degraded')
def test_retrieval_diversification_is_bounded_and_keeps_anaphora_context(self):
history = [{'role': 'user', 'content': 'Carbon inventory handoff'}]
current = 'Thanks. That still needs a clear owner. I cannot verify its source.'
queries = recall.retrieval_queries(current, history)
self.assertEqual(len(queries), 3)
self.assertEqual(queries[0], current)
self.assertIn('Carbon inventory handoff', queries[1])
self.assertIn('I cannot verify its source.', queries)
self.assertEqual(recall.retrieval_queries('A single current need', []), ['A single current need'])
self.assertEqual(recall.retrieval_queries('Finish it. OK', []), ['Finish it. OK'])
def test_cold_parallel_vector_reads_never_observe_unfinished_cache(self):
from concurrent.futures import ThreadPoolExecutor
import array
meta = self.root / 'semantic.json'
meta.write_text(json.dumps({'dims': 2, 'keys': ['wiki/topics/test.md']}), encoding='utf-8')
meta.with_suffix('.bin').write_bytes(array.array('f', [1, 0]).tobytes())
entered, release = threading.Event(), threading.Event()
real_load = json.load
def slow_load(file):
entered.set()
self.assertTrue(release.wait(5))
return real_load(file)
with mock.patch.dict('os.environ', {'XKB_SEMANTIC_INDEX': str(meta)}), \
mock.patch.object(cr, '_VECTORS', None), \
mock.patch.object(cr.json, 'load', side_effect=slow_load), \
ThreadPoolExecutor(max_workers=2) as pool:
first = pool.submit(cr._load_semantic_vectors)
self.assertTrue(entered.wait(5))
second = pool.submit(cr._load_semantic_vectors)
try:
self.assertFalse(second.done())
finally:
release.set()
self.assertEqual(first.result(), {'wiki/topics/test.md': [1, 0]})
self.assertEqual(second.result(), first.result())
def test_parallel_catalog_layers_preserve_both_acl_diagnostics(self):
catalog = service.KnowledgeCatalog()
barrier = threading.Barrier(2)
def search(layer, query, limit, namespace):
catalog._stats = service.filter_stats(**({'semantic': 2} if layer == 'cards' else {'wiki': 3}))
catalog._irrelevant = 1 if layer == 'cards' else 0
barrier.wait(timeout=5)
return [{'id': layer, 'score': .8, 'namespace': namespace}]
with mock.patch.object(catalog, '_semantic_search', side_effect=lambda *a: search('cards', *a)), \
mock.patch.object(catalog, '_wiki_search', side_effect=lambda *a: search('wiki', *a)):
packet = catalog.search('two independent reads', namespace='client-a')
self.assertEqual(packet['filtered_counts']['total'], 5)
self.assertEqual(packet['filtered_counts']['by_layer']['semantic'], 2)
self.assertEqual(packet['filtered_counts']['by_layer']['wiki'], 3)
self.assertEqual(packet['dropped_as_irrelevant'], 1)
self.assertEqual({r['namespace'] for r in packet['records']}, {'client-a'})
self.assertTrue(all(s['used'] for s in packet['backends'].values()))
def test_judge_retains_all_bounded_context_within_body_limit(self):
history = [{"role": "user", "content": "offline deployment only" + "x" * 570},
{"role": "assistant", "content": "y" * 600}] * 2
query = recall.contextual_query("z" * 1200, history, for_judge=True)
def judge(state, questions, **kwargs):
self.assertEqual(state.count("offline deployment only"), 2)
self.assertLessEqual(len(jev._body(state, questions)), jev.MAX_BODY_BYTES)
return {key: {"noul": 1} for key in questions}
with mock.patch.object(jev, "judge", side_effect=judge):
self.assertTrue(jev.relevance(query, [(str(n), "candidate" * 300) for n in range(25)]))
def test_context_merge_preserves_ranking_legs_and_anonymous_evidence(self):
store = service.Store(self.root / "knowledge.sqlite")
def search(query, *args, **kwargs):
if "earlier" in query:
return {"records": [{"id": "a", "score": 1, "score_scale": "card_keyword"}]}
return {"records": [{"id": "a", "score": .8, "score_scale": "card_semantic"},
{"id": "b", "score": .9, "score_scale": "card_semantic"},
{"summary": "anonymous one", "score": .1},
{"summary": "anonymous two", "score": .1}]}
with mock.patch.object(store.catalog, "search", side_effect=search), \
mock.patch.object(store, "recall", return_value={"memories": []}), \
mock.patch.dict("os.environ", {"XKB_JEV_DECIDE": "0"}):
packet = store.knowledge_recall("current need", 5,
conversation=[{"role": "user", "content": "earlier context"}])
self.assertEqual(packet['records'][0]['id'], 'a')
self.assertEqual(set(packet['records'][0]['matched_by']), {'card_semantic', 'card_keyword'})
self.assertEqual(len(packet['records']), 4)
class TopicResolution(unittest.TestCase):
def test_resolution_keeps_source_and_identity_and_rejects_stale_content(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
staging = root / "staging"
staging.mkdir()
governance = root / "governance"
governance.mkdir()
source = staging / "sample.md"
original = "## Candidate 1\n- **Topic:** old-topic\n- **Status:** [ ] approve\n\nReusable observation with evidence.\n"
source.write_text(original, encoding="utf-8")
with mock.patch.object(review, "STAGING_DIR", staging), mock.patch.object(review, "GOVERNANCE_DIR", governance):
before = review.load_candidates(False)[0]
(governance / "topic-resolutions.json").write_text(json.dumps({before.candidate_id: {
"fingerprint": before.fingerprint, "topic": "current-topic", "reason": "reviewed mapping"}}), encoding="utf-8")
after = review.load_candidates(False)[0]
self.assertEqual(after.topic_key, "current-topic")
self.assertEqual(after.original_topic, "old-topic")
self.assertEqual(after.candidate_id, before.candidate_id)
self.assertEqual(source.read_text(encoding="utf-8"), original)
topics = root / "topics"
topics.mkdir()
(topics / "current-topic.md").write_text("# Current\n", encoding="utf-8")
with mock.patch.object(distill, "STAGING_DIR", staging), \
mock.patch.object(distill, "TOPICS_DIR", topics):
applied, skipped, slugs = distill.apply_staging_file(source, approve_all=True)
self.assertEqual((applied, skipped, slugs), (1, 0, ["current-topic"]))
self.assertFalse((topics / "old-topic.md").exists())
source.write_text(original + "Changed evidence.\n", encoding="utf-8")
with self.assertRaisesRegex(ValueError, "stale topic resolution"):
review.load_candidates(False)
def test_archived_and_redirect_pages_cannot_accumulate_new_candidates(self):
with tempfile.TemporaryDirectory() as tmp:
topics = Path(tmp)
for name, body in {"active": "# Active\n", "archive": "---\nstatus: archived\n---\n# Archive\n",
"redirect": "# Redirect: Canonical\n[[active]]\n"}.items():
(topics / f"{name}.md").write_text(body, encoding="utf-8")
with mock.patch.object(distill, "TOPICS_DIR", topics):
self.assertEqual(distill.load_topic_slugs(), ["active"])
before = (topics / "redirect.md").read_bytes()
self.assertFalse(distill.upsert_wiki_section("redirect", "Notes", "New note", "2026-10-05"))
self.assertEqual((topics / "redirect.md").read_bytes(), before)
if __name__ == "__main__":
unittest.main()