agentic.email/backend/tests/test_human_handoff_service.py

import pytest

from app.services.human_handoff_service import (
    MESSAGE_SCOPE_EXTERNAL,
    MESSAGE_SCOPE_HANDOFF,
    _body_has_quote_trail,
    _mirror_from_identity,
    build_handoff_subject,
    maybe_trigger_handoff,
    parse_human_initiated_subject,
    process_human_reply,
    redact_human_address,
    resolve_handoff_from_inbound,
    select_matching_handoff_rule,
)


def test_scope_constants_are_distinct():
    assert MESSAGE_SCOPE_EXTERNAL != MESSAGE_SCOPE_HANDOFF


def test_human_handoff_runtime_is_enabled():
    assert callable(maybe_trigger_handoff)
    assert callable(process_human_reply)
    assert callable(resolve_handoff_from_inbound)


@pytest.mark.asyncio
async def test_mirror_body_appends_external_history_when_missing(monkeypatch):
    from types import SimpleNamespace
    from unittest.mock import AsyncMock

    from app.services.human_handoff_service import _mirror_body_with_trail

    parent = {
        "_id": "parent",
        "sender": {"name": "Parth Dixit", "address": "parthdixit091@gmail.com"},
        "body_text": "sender: C\n\nOn earlier Suyash Dixit wrote:\n> responder: B",
        "body_html": None,
        "sent_at": None,
    }
    monkeypatch.setattr(
        "app.services.human_handoff_service._latest_thread_message",
        AsyncMock(return_value=parent),
    )
    monkeypatch.setattr(
        "app.services.human_handoff_service.load_mirror_payload",
        AsyncMock(return_value=SimpleNamespace(body_text=parent["body_text"], body_html=None)),
    )
    text, html = await _mirror_body_with_trail(
        SimpleNamespace(),
        account_id="account-1",
        handoff={"external_thread_id": "thread"},
        source={"_id": "child", "new_reply_content": "responder: D", "new_reply_html": None},
        payload=SimpleNamespace(body_text="responder: D", body_html=None),
    )
    assert text.startswith("responder: D")
    assert "parthdixit091@gmail.com" in text
    assert "> sender: C" in text
    assert "responder: B" in text
    assert html and "gmail_quote" in html


def test_quote_trail_is_detected_in_forwarded_history():
    assert _body_has_quote_trail(
        "responder: B\n\nOn Tue, Oct 6, 2026 at 7:44 AM Suyash Dixit wrote:\n> sender: A",
        None,
    )
    assert _body_has_quote_trail(
        "sender: C",
        '<div>sender: C</div><div class="gmail_quote">sender: A</div>',
    )
    assert _body_has_quote_trail("sender: E", None) is False


def test_mirror_from_stays_on_gentic_address():
    name, address = _mirror_from_identity(
        {
            "direction": "inbound",
            "sender": {
                "name": "Parth Dixit",
                "address": "parthdixit091@gmail.com",
            },
        },
        "Suyash Dixit",
        "suyash@gentic.email",
    )
    assert name == "Parth Dixit"
    assert address == "suyash@gentic.email"
    assert "gmail.com" not in address

    agent_name, agent_address = _mirror_from_identity(
        {"direction": "outbound", "sender": {"address": "suyash@gentic.email"}},
        "Suyash Dixit",
        "suyash@gentic.email",
    )
    assert agent_name == "Suyash Dixit"
    assert agent_address == "suyash@gentic.email"


def test_human_initiated_subject_accepts_dot_hyphen_and_bar():
    assert parse_human_initiated_subject(
        "handoff · Suyash Dixit · suyashdixit091@gmail.com · hello"
    ) == {
        "customer_name": "Suyash Dixit",
        "customer_email": "suyashdixit091@gmail.com",
        "subject": "hello",
    }
    assert parse_human_initiated_subject(
        "handoff -  Suyash Dixit  - suyashdixit091@gmail.com - Need a quote"
    )["subject"] == "Need a quote"
    parsed = parse_human_initiated_subject(
        "Re: handoff | Suyash Dixit | SuyashDixit091@gmail.com | hello - there"
    )
    assert parsed["customer_email"] == "suyashdixit091@gmail.com"
    assert parsed["customer_name"] == "Suyash Dixit"
    assert parsed["subject"] == "hello - there"


def test_human_initiated_subject_rejects_incomplete_patterns():
    assert parse_human_initiated_subject("Suyash Dixit · handoff · a@b.com · hello") is None
    assert parse_human_initiated_subject("handoff · Suyash Dixit · not-an-email · hello") is None
    assert parse_human_initiated_subject("handoff-Suyash Dixit-a@b.com-hello") is None
    assert parse_human_initiated_subject("hello from the customer") is None


def test_private_subject_contains_customer_name_and_external_subject():
    subject = build_handoff_subject(
        sender_name="Parth Dixit",
        external_sender="parthdixit091@gmail.com",
        external_subject="Invoice 42",
    )
    assert subject == (
        "handoff · Parth Dixit · parthdixit091@gmail.com · Invoice 42"
    )


def test_private_subject_removes_header_line_breaks():
    subject = build_handoff_subject(
        sender_name="Urgent\r\nBcc: attacker@example.com",
        external_sender="customer@example.com",
        external_subject="Help",
    )
    assert "\r" not in subject
    assert "\n" not in subject


def test_human_address_is_replaced_before_external_relay():
    value = "Regards,\nOwner <private.owner@example.com>"
    assert redact_human_address(
        value,
        human_email="private.owner@example.com",
        agent_email="agent@gentic.email",
    ) == "Regards,\nOwner <agent@gentic.email>"


def test_lowest_priority_matching_rule_wins():
    rules = [
        {
            "name": "Backup",
            "priority": 20,
            "labels": ["All mail"],
            "email_types": ["all"],
            "instructions": "",
        },
        {
            "name": "Primary",
            "priority": 1,
            "labels": ["All mail"],
            "email_types": ["all"],
            "instructions": "",
        },
    ]
    selected = select_matching_handoff_rule(
        rules,
        email={"subject": "Hello", "body_text": "Need help"},
    )
    assert selected["name"] == "Primary"