#!/usr/bin/env python3 """Focused bugcheck for Customer Agent domain + package consistency.""" from __future__ import annotations import importlib.util import sqlite3 import sys import tempfile from pathlib import Path def load_ca(): path = Path(__file__).resolve().parents[2] / "scripts" / "customer_agent.py" spec = importlib.util.spec_from_file_location("ca_bugcheck", path) assert spec and spec.loader mod = importlib.util.module_from_spec(spec) sys.modules[spec.name] = mod spec.loader.exec_module(mod) return mod def main() -> int: ca = load_ca() failures: list[str] = [] def check(cond: bool, msg: str) -> None: if not cond: failures.append(msg) print(f"FAIL {msg}") else: print(f"PASS {msg}") # Brand validation check(ca.normalize_brand_color("") == "", "empty brand color ok") check(ca.normalize_brand_color("#2e4a86") == "#2e4a86", "hex color ok") try: ca.normalize_brand_color("navy") check(False, "invalid color rejected") except ValueError: check(True, "invalid color rejected") # Origins check(ca.origin_allowed(["*"], "https://a.com"), "star allows") check(not ca.origin_allowed(["https://a.com"], "https://b.com"), "deny foreign") check(ca.origin_allowed(["https://Shop.Example.com"], "https://shop.example.com"), "origin case normalize") # Retrieval chunks = [ {"title": "Shipping", "text": "Free shipping over EUR 60 across the EU in 3-5 days."}, {"title": "Hours", "text": "Open Monday to Saturday 9-18."}, ] hits = ca.retrieve_chunks(chunks, "free shipping to Germany") check(bool(hits) and "ship" in hits[0]["title"].casefold(), "shipping ranks first") check(ca.knowledge_is_hit(hits), "shipping is knowledge hit") miss = ca.retrieve_chunks(chunks, "wifi password for cafe guests") check(not ca.knowledge_is_hit(miss) or (miss and miss[0].get("score", 0) < 0.35), "wifi is not a solid hit") # Guardrails check(ca.looks_hard_refuse("diagnose my illness"), "hard refuse medical") check(ca.looks_sensitive("I want to speak to a human"), "sensitive human") ans = ca.hard_refuse_answer("help@x.com") check("cannot help" in ans.casefold() or "cannot" in ans.casefold(), "hard refuse text") # Schema + create roundtrip with tempfile.TemporaryDirectory() as tmp: db = Path(tmp) / "t.sqlite3" conn = sqlite3.connect(db) conn.row_factory = sqlite3.Row conn.execute("CREATE TABLE registered_users (customer_id TEXT PRIMARY KEY, email TEXT NOT NULL)") conn.execute("INSERT INTO registered_users(customer_id, email) VALUES (?, ?)", ("c1", "a@example.com")) ca.ensure_customer_agent_schema(conn) agent, key = ca.create_agent( conn, customer_id="c1", name="Shop", escalate_email="help@example.com", allowed_origins=["*"], brand_position="left", brand_primary_color="#112233", ) check(key.startswith("pk_live_"), "site key prefix") check(agent.brand_position == "left", "brand position stored") loaded = ca.get_agent_by_site_key(conn, key) check(loaded is not None and loaded.name == "Shop", "lookup by site key") try: ca.add_knowledge_doc(conn, agent_id=agent.id, title="x", body=" ") check(False, "empty knowledge rejected") except ValueError: check(True, "empty knowledge rejected") doc = ca.add_knowledge_doc(conn, agent_id=agent.id, title="FAQ", body="Open 9-5 Monday to Friday.") check(doc["chars"] > 0, "knowledge added") try: ca.update_agent(conn, customer_id="c1", agent_id=agent.id, escalate_webhook_url="http://x.com") check(False, "http webhook rejected") except ValueError: check(True, "http webhook rejected") # Use a public host that resolves; webhook delivery may still fail HTTP-wise. ca.update_agent( conn, customer_id="c1", agent_id=agent.id, escalate_webhook_url="https://example.com/hooks/synderesis", allowed_origins=["https://shop.example.com"], ) updated = ca.get_agent_for_customer(conn, "c1", agent.id) assert updated is not None ready = ca.readiness_checklist(updated, knowledge_count=1) check(ready["ready_for_production"] is True, "ready for production when locked + FAQ + email") # URL validation try: ca.validate_public_https_url("http://example.com") check(False, "http url rejected") except ValueError: check(True, "http url rejected") try: ca.validate_public_https_url("https://127.0.0.1/x") check(False, "loopback url rejected") except ValueError: check(True, "loopback url rejected") # Escalation + delivery skip without SMTP esc = ca.create_escalation( conn, agent_id=agent.id, visitor_name="A", visitor_email="a@b.com", message="Help", transcript=[{"role": "user", "content": "hi"}], ) delivery = ca.deliver_escalation_notifications( agent=updated, escalation={**esc, "visitor_name": "A", "visitor_email": "a@b.com", "message": "Help", "transcript": []}, mail=ca.EscalationMailConfig(enabled=False), webhook_signing_secret="secret", ) check(delivery["email_status"] == "skipped", "email skipped when smtp off") # webhook to example.com may fail DNS/HTTP - status should be sent or failed not empty check(delivery["webhook_status"] in ("sent", "failed:http_404", "failed:http_403") or delivery["webhook_status"].startswith("failed:"), "webhook attempted") # Chat event + coverage ca.log_chat_event( conn, agent_id=agent.id, question="wifi password?", answer="I don't know", knowledge_hit=False, escalate_recommended=True, hard_refuse=False, retrieved_count=0, top_score=0.0, ) cov = ca.coverage_report(conn, agent.id) check(cov["total_chats"] == 1, "coverage counts chats") check(len(cov["unknown_questions"]) >= 1, "unknown questions listed") # Export bundle = ca.export_agent_bundle(conn, updated) check(bundle["object"].endswith("export"), "export object") check("site_key" not in bundle["agent"], "export omits raw site key") conn.close() # Package consistency root = Path(__file__).resolve().parents[1] version = (root / "VERSION").read_text(encoding="utf-8").strip() check(bool(version) and version[0].isdigit(), f"VERSION present (got {version})") php = (root / "integrations/wordpress/synderesis-customer-agent/synderesis-customer-agent.php").read_text(encoding="utf-8") check("1.4.0" in php, "WP plugin version present") check("synderesis_ca_test_connection" in php, "WP has test connection") check((root / "integrations/wordpress/synderesis-customer-agent/uninstall.php").is_file(), "uninstall.php exists") if failures: print(f"\n{len(failures)} failure(s)") return 1 print("\nBUGCHECK PASSED") return 0 if __name__ == "__main__": raise SystemExit(main())