customer-agent/scripts/bugcheck.py

#!/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())