"""Empirical Adversarial Test Harness: Semantic Stress Testing & Query Probing. Milestone 3 — Teamwork Preview Challenger 1 (teamwork_preview_challenger_m3_1) Project: DSCons ERP Mem0 Long-Term Knowledge Base Overhaul Comprehensive adversarial testing of Mem0 long-term memory engine across: 1. Category 1: Vietnamese Natural Domain Queries (10 queries) 2. Category 2: English Semantic Variations (10 queries) 3. Category 3: Typos, Accents, Case Variations (8 queries) 4. Category 4: Long Multi-Clause Engineering Prompts (6 queries) Evaluates: - Mode A: Unscoped Search (Default global search across developer memories) - Mode B: Scoped Search (Filtered with project='DSCons') - Cosine similarity scores (Score >= 0.55 threshold) - Domain dominance (Top-1 is DSCons, Top-3 >= 66% DSCons) - Contamination (FreeExile, VoLamWeb, legacy DSCons strings) - Keyword & target fact relevance """ import asyncio import json import re import sys from typing import Any, Optional from mcp.client.session import ClientSession from mcp.client.sse import sse_client if hasattr(sys.stdout, "reconfigure"): sys.stdout.reconfigure(encoding="utf-8", errors="replace") if hasattr(sys.stderr, "reconfigure"): sys.stderr.reconfigure(encoding="utf-8", errors="replace") MEM0_SSE_URL = "http://127.0.0.1:8765/sse" USER_ID = "developer" MIN_SIMILARITY_SCORE = 0.55 FORBIDDEN_LEGACY_STRINGS = [ "cache 71 hạng mục", "4.753 tỷ", "app/core/logging_config", ] FOREIGN_KEYWORDS = [ "freeexile", "feral npc", "dialogue hud", "combat ui", "savage infinite atlas", "cocos creator", "volamweb", ] # Adversarial test queries catalog TEST_SUITE = [ # ------------------------------------------------------------- # Category 1: Vietnamese Domain Queries (Mandatory variations) # ------------------------------------------------------------- { "id": "VN-01", "category": "Vietnamese Domain", "query": "quy tắc DSCons", "target_area": "Area 1: Root Standards", "target_keywords": ["điều luật", "định sơn", "nguyên tắc", "luật", "dscons"], }, { "id": "VN-02", "category": "Vietnamese Domain", "query": "luật bất biến DSCons", "target_area": "Area 1: Root Standards", "target_keywords": ["bảy điều luật", "bất biến", "định sơn", "0202111150"], }, { "id": "VN-03", "category": "Vietnamese Domain", "query": "giao diện dark slate ERP", "target_area": "Area 3: Frontend UI", "target_keywords": ["dark slate", "theme", "bloomberg", "#0b0f19"], }, { "id": "VN-04", "category": "Vietnamese Domain", "query": "bảng màu frontend", "target_area": "Area 3: Frontend UI", "target_keywords": ["dark slate", "màu", "palette", "#0b0f19", "#080d1a"], }, { "id": "VN-05", "category": "Vietnamese Domain", "query": "kiến trúc backend lục giác", "target_area": "Area 2: Backend & DB", "target_keywords": ["lục giác", "clean architecture", "ddd", "ports"], }, { "id": "VN-06", "category": "Vietnamese Domain", "query": "độ chính xác số thực database", "target_area": "Area 2: Backend & DB", "target_keywords": ["numeric", "18, 4", "decimal", "sổ kép"], }, { "id": "VN-07", "category": "Vietnamese Domain", "query": "định mức cừ larsen VL=0", "target_area": "Area 4: AEC Business", "target_keywords": ["larsen", "thông tư 38", "vl = 0", "vl=0", "định mức"], }, { "id": "VN-08", "category": "Vietnamese Domain", "query": "tỷ trọng MR/T", "target_area": "Area 4: AEC Business", "target_keywords": ["mr/t", "tỷ trọng", "vật tư", "40-65%"], }, { "id": "VN-09", "category": "Vietnamese Domain", "query": "bóc tách CAD TCVN3", "target_area": "Area 4: AEC Business", "target_keywords": ["tcvn3", "cad", "bóc tách", "bounding box"], }, { "id": "VN-10", "category": "Vietnamese Domain", "query": "deep matrix task.md", "target_area": "Area 5: Testing & Workflow", "target_keywords": ["deep matrix", "task.md", "d1", "d6", "kế hoạch"], }, # ------------------------------------------------------------- # Category 2: English Semantic Variations # ------------------------------------------------------------- { "id": "EN-01", "category": "English Semantic", "query": "DSCons core rules", "target_area": "Area 1: Root Standards", "target_keywords": ["rule", "điều luật", "định sơn", "dscons"], }, { "id": "EN-02", "category": "English Semantic", "query": "immutable laws DSCons", "target_area": "Area 1: Root Standards", "target_keywords": ["điều luật", "bất biến", "định sơn"], }, { "id": "EN-03", "category": "English Semantic", "query": "dark slate ERP UI", "target_area": "Area 3: Frontend UI", "target_keywords": ["dark slate", "theme", "bloomberg", "erp"], }, { "id": "EN-04", "category": "English Semantic", "query": "frontend color palette", "target_area": "Area 3: Frontend UI", "target_keywords": ["dark slate", "palette", "#0b0f19"], }, { "id": "EN-05", "category": "English Semantic", "query": "hexagonal backend architecture", "target_area": "Area 2: Backend & DB", "target_keywords": ["hexagonal", "lục giác", "clean architecture", "ports"], }, { "id": "EN-06", "category": "English Semantic", "query": "database decimal precision", "target_area": "Area 2: Backend & DB", "target_keywords": ["numeric", "decimal", "precision", "18, 4"], }, { "id": "EN-07", "category": "English Semantic", "query": "Larsen sheet pile circular 38 VL=0", "target_area": "Area 4: AEC Business", "target_keywords": ["larsen", "circular 38", "thông tư 38", "vl = 0", "vl=0"], }, { "id": "EN-08", "category": "English Semantic", "query": "material budget ratio MR/T", "target_area": "Area 4: AEC Business", "target_keywords": ["mr/t", "mr / t", "tỷ trọng", "vật tư", "budget"], }, { "id": "EN-09", "category": "English Semantic", "query": "CAD drawing takeoff TCVN3", "target_area": "Area 4: AEC Business", "target_keywords": ["cad", "tcvn3", "takeoff", "bóc tách"], }, { "id": "EN-10", "category": "English Semantic", "query": "deep matrix planning workflow", "target_area": "Area 5: Testing & Workflow", "target_keywords": ["deep matrix", "workflow", "d1", "task.md"], }, # ------------------------------------------------------------- # Category 3: Edge Conditions (Typos, Case, Accents) # ------------------------------------------------------------- { "id": "EDGE-TYPO-01", "category": "Typos & Accents", "query": "quy tac DSCOns", "target_area": "Area 1: Root Standards", "target_keywords": ["điều luật", "định sơn", "dscons"], }, { "id": "EDGE-TYPO-02", "category": "Typos & Accents", "query": "giaodien darkslate ERP", "target_area": "Area 3: Frontend UI", "target_keywords": ["dark slate", "theme", "erp"], }, { "id": "EDGE-TYPO-03", "category": "Typos & Accents", "query": "kien truc backend luc giac", "target_area": "Area 2: Backend & DB", "target_keywords": ["clean architecture", "lục giác", "ports"], }, { "id": "EDGE-TYPO-04", "category": "Typos & Accents", "query": "dinh muc cu larsen VL=0", "target_area": "Area 4: AEC Business", "target_keywords": ["larsen", "thông tư 38", "vl = 0", "định mức"], }, { "id": "EDGE-TYPO-05", "category": "Typos & Accents", "query": "boc tach CAD TCVN3", "target_area": "Area 4: AEC Business", "target_keywords": ["cad", "tcvn3", "bóc tách"], }, { "id": "EDGE-CASE-01", "category": "Case Sensitivity", "query": "DSCONS RULE", "target_area": "Area 1: Root Standards", "target_keywords": ["điều luật", "định sơn", "dscons"], }, { "id": "EDGE-CASE-02", "category": "Case Sensitivity", "query": "QuY TắC DsCoNs", "target_area": "Area 1: Root Standards", "target_keywords": ["điều luật", "định sơn", "dscons"], }, { "id": "EDGE-CASE-03", "category": "Case Sensitivity", "query": "FRONTEND THEME DARK SLATE", "target_area": "Area 3: Frontend UI", "target_keywords": ["dark slate", "theme", "#0b0f19"], }, # ------------------------------------------------------------- # Category 4: Long Multi-Clause Engineering Prompts # ------------------------------------------------------------- { "id": "LONG-01", "category": "Long Prompts", "query": "Tôi muốn biết quy tắc bất biến về ranh giới kiến trúc lục giác DDD và cấm raw SQL trong router của DSCons ERP", "target_area": "Area 2: Backend & DB", "target_keywords": ["lục giác", "hexagonal", "ddd", "raw sql", "router"], }, { "id": "LONG-02", "category": "Long Prompts", "query": "Quy định về việc sử dụng kiểu dữ liệu NUMERIC(18, 4) và nguyên tắc ghi sổ kép cho tài chính kế toán tại DSCons", "target_area": "Area 2: Backend & DB", "target_keywords": ["numeric", "18, 4", "sổ kép", "tài chính"], }, { "id": "LONG-03", "category": "Long Prompts", "query": "Cho tôi biết hướng dẫn thiết kế giao diện Dark Slate và màu sắc chuẩn của công ty Định Sơn", "target_area": "Area 3: Frontend UI", "target_keywords": ["dark slate", "theme", "#0b0f19", "định sơn"], }, { "id": "LONG-04", "category": "Long Prompts", "query": "Cách tính định mức cừ Larsen theo Thông tư 38 và tỷ trọng ngân sách vật tư MR trên T", "target_area": "Area 4: AEC Business", "target_keywords": ["larsen", "thông tư 38", "mr/t", "vật tư"], }, { "id": "LONG-05", "category": "Long Prompts", "query": "Quy chuẩn bóc tách bản vẽ kỹ thuật CAD giải mã phông chữ tiếng Việt TCVN3 và cắt khung Bounding Box", "target_area": "Area 4: AEC Business", "target_keywords": ["cad", "tcvn3", "bounding box", "bóc tách"], }, { "id": "LONG-06", "category": "Long Prompts", "query": "Quy trình lập kế hoạch đa tầng Deep Matrix D1 đến D6 trên task.md và giao thức TDD khép kín", "target_area": "Area 5: Testing & Workflow", "target_keywords": ["deep matrix", "task.md", "d1", "tdd"], }, ] def is_dscons(item: dict[str, Any]) -> bool: text = item.get("memory", "") metadata = item.get("metadata") or {} proj = str(metadata.get("project", "")).strip().lower() if proj == "dscons": return True return bool(text.startswith("[DSCons") or "DSCons ERP" in text or "Định Sơn" in text) def detect_foreign_or_legacy(text: str) -> list[str]: violations = [] text_lower = text.lower() for legacy in FORBIDDEN_LEGACY_STRINGS: if legacy.lower() in text_lower: violations.append(f"Legacy string: '{legacy}'") for foreign in FOREIGN_KEYWORDS: if foreign in text_lower: violations.append(f"Foreign contamination: '{foreign}'") return violations async def evaluate_query(session: ClientSession, test: dict[str, Any], project_scope: Optional[str] = None) -> dict[str, Any]: args = {"query": test["query"], "user_id": USER_ID, "limit": 5} if project_scope: args["project"] = project_scope search_resp = await session.call_tool("mem0_search", arguments=args) data = json.loads(search_resp.content[0].text) hits = data.get("memories", {}).get("results", []) top1_score = 0.0 top1_is_dscons = False top1_text = "" top3_dscons_count = 0 contaminations = [] kw_matched = False if hits: top1 = hits[0] top1_score = float(top1.get("score", 0.0)) top1_text = top1.get("memory", "") top1_is_dscons = is_dscons(top1) top3 = hits[:3] top3_dscons_count = sum(1 for h in top3 if is_dscons(h)) for idx, h in enumerate(top3): h_text = h.get("memory", "") violations = detect_foreign_or_legacy(h_text) if violations: contaminations.append(f"Hit #{idx+1}: {', '.join(violations)}") for kw in test["target_keywords"]: if kw.lower() in top1_text.lower(): kw_matched = True break score_pass = top1_score >= MIN_SIMILARITY_SCORE dom_pass = top1_is_dscons and (top3_dscons_count >= 2) contam_clean = len(contaminations) == 0 overall_pass = score_pass and dom_pass and contam_clean and kw_matched return { "test": test, "hits": hits, "top1_score": top1_score, "top1_is_dscons": top1_is_dscons, "top1_text": top1_text, "top3_dscons_count": top3_dscons_count, "contaminations": contaminations, "kw_matched": kw_matched, "score_pass": score_pass, "dom_pass": dom_pass, "contam_clean": contam_clean, "overall_pass": overall_pass, } async def run_comparative_suite(): print("=" * 115) print(" DSCONS ERP MEM0 ADVERSARIAL SEMANTIC STRESS TEST SUITE (CHALLENGER 1)") print(" COMPARING: UNSCOPED SEARCH vs SCOPED SEARCH (project='DSCons')") print("=" * 115) async with ( sse_client(MEM0_SSE_URL) as (read, write), ClientSession(read, write) as session, ): await session.initialize() status_resp = await session.call_tool("mem0_status", arguments={}) status_data = json.loads(status_resp.content[0].text) print(f"Service status: {status_data.get('status')} | Embedder: {status_data.get('embedder')}") print(f"Total memories in Qdrant: {status_data.get('developer_memories_count')}") print("-" * 115) unscoped_results = [] scoped_results = [] print(f"{'ID':<14} | {'Query':<30} | {'Unscoped Top-1':<14} | {'Unscoped':<8} | {'Scoped Top-1':<12} | {'Scoped':<6}") print("-" * 115) for test in TEST_SUITE: res_u = await evaluate_query(session, test, project_scope=None) res_s = await evaluate_query(session, test, project_scope="DSCons") unscoped_results.append(res_u) scoped_results.append(res_s) u_verdict = "PASS" if res_u["overall_pass"] else "FAIL" s_verdict = "PASS" if res_s["overall_pass"] else "FAIL" u_score_str = f"{res_u['top1_score']:.4f} ({'DSC' if res_u['top1_is_dscons'] else 'FOR'})" s_score_str = f"{res_s['top1_score']:.4f} ({'DSC' if res_s['top1_is_dscons'] else 'FOR'})" short_q = (test["query"][:28] + "..") if len(test["query"]) > 28 else test["query"] print(f"{test['id']:<14} | {short_q:<30} | {u_score_str:<14} | {u_verdict:<8} | {s_score_str:<12} | {s_verdict:<6}") # Aggregates total = len(TEST_SUITE) u_passed = sum(1 for r in unscoped_results if r["overall_pass"]) s_passed = sum(1 for r in scoped_results if r["overall_pass"]) print("=" * 115) print(f"COMPARATIVE SUMMARY:") print(f" MODE A - Unscoped Search (Default global): {u_passed}/{total} Passed ({(u_passed/total)*100:.1f}%)") print(f" MODE B - Scoped Search (project='DSCons'): {s_passed}/{total} Passed ({(s_passed/total)*100:.1f}%)") print("=" * 115) # In-depth breakdown for Mode B failures s_failures = [r for r in scoped_results if not r["overall_pass"]] print(f"\nRemaining Failures even under Scoped Search ({len(s_failures)}/{total}):") for idx, f in enumerate(s_failures, 1): t = f["test"] print(f"\n [Scoped Failure #{idx}] {t['id']}: '{t['query']}'") print(f" Top-1 Score: {f['top1_score']:.4f} (>= 0.55: {f['score_pass']})") print(f" Top-1 Is DSCons: {f['top1_is_dscons']}") print(f" Top-3 DSCons Count: {f['top3_dscons_count']}/3 (>= 2: {f['dom_pass']})") print(f" Keyword Matched: {f['kw_matched']} (Target: {t['target_keywords']})") print(f" Contaminations: {f['contaminations']}") print(f" Top-1 Text Snippet: {f['top1_text'][:140]}...") # Write comparative JSON report report = { "total_queries": total, "unscoped_passed": u_passed, "unscoped_pass_rate": (u_passed / total) * 100.0, "scoped_passed": s_passed, "scoped_pass_rate": (s_passed / total) * 100.0, "unscoped_results": [ { "id": r["test"]["id"], "query": r["test"]["query"], "category": r["test"]["category"], "top1_score": r["top1_score"], "top1_is_dscons": r["top1_is_dscons"], "top3_dscons_count": r["top3_dscons_count"], "kw_matched": r["kw_matched"], "contaminations": r["contaminations"], "pass": r["overall_pass"], "snippet": r["top1_text"][:120], } for r in unscoped_results ], "scoped_results": [ { "id": r["test"]["id"], "query": r["test"]["query"], "category": r["test"]["category"], "top1_score": r["top1_score"], "top1_is_dscons": r["top1_is_dscons"], "top3_dscons_count": r["top3_dscons_count"], "kw_matched": r["kw_matched"], "contaminations": r["contaminations"], "pass": r["overall_pass"], "snippet": r["top1_text"][:120], } for r in scoped_results ], } with open("reports_adversarial_comparative.json", "w", encoding="utf-8") as fp: json.dump(report, fp, ensure_ascii=False, indent=2) print("\nSaved comparative report to reports_adversarial_comparative.json") if __name__ == "__main__": asyncio.run(run_comparative_suite())