salespath_env/server/rules.py

# salespath_env/server/rules.py

from dataclasses import dataclass
from typing import Callable

from ..models import SalesPathAction, SalesPathState


@dataclass
class BusinessRule:
    """
    Returns True when the rule is VIOLATED.
    """

    rule_id: str
    name: str
    description: str
    check: Callable[[SalesPathState, SalesPathAction], bool]


def _qualify_before_present(
    state: SalesPathState,
    action: SalesPathAction,
) -> bool:
    """
    R01:
    PRESENT before QUALIFY is invalid.
    """
    if action.action_type == "PRESENT":
        return "QUALIFY" not in state.steps_completed
    return False


def _demo_before_negotiate(
    state: SalesPathState,
    action: SalesPathAction,
) -> bool:
    """
    R02:
    NEGOTIATE before OFFER_DEMO is invalid.
    """
    if action.action_type == "NEGOTIATE":
        return "OFFER_DEMO" not in state.steps_completed
    return False


def _budget_known_to_negotiate(
    state: SalesPathState,
    action: SalesPathAction,
) -> bool:
    """
    R03:
    Cannot NEGOTIATE while budget is unknown.
    """
    if action.action_type == "NEGOTIATE":
        return state.prospect_profile.get("budget_signal") == "unknown"
    return False


def _discount_after_objections(
    state: SalesPathState,
    action: SalesPathAction,
) -> bool:
    """
    R04:
    Discount only after 2 objections handled.
    """
    if action.action_type == "NEGOTIATE":
        if "discount" in action.content.lower():
            return state.objections_handled < 2
    return False


def _no_repeat_action(
    state: SalesPathState,
    action: SalesPathAction,
) -> bool:
    """
    R05:
    Same action twice in a row is invalid.
    FIX: conversation_history alternates agent/prospect entries.
    Must filter to agent-only turns before comparing.
    """
    agent_turns = [
        e for e in state.conversation_history
        if e.get("speaker") == "agent"
    ]
    if agent_turns:
        return agent_turns[-1].get("action_type", "") == action.action_type
    return False


def _prospect_first(
    state: SalesPathState,
    action: SalesPathAction,
) -> bool:
    """
    R06:
    First action must be PROSPECT.
    """
    if state.turn_number == 1:
        return action.action_type != "PROSPECT"
    return False


def _followup_timing(
    state: SalesPathState,
    action: SalesPathAction,
) -> bool:
    """
    R07:
    FOLLOW_UP only valid after prospect silence (no response for 1+ agent turns).
    Violation if the prospect HAS replied since the last agent action.
    FIX: Previous logic was inverted — it was blocking valid FOLLOW_UP.
    """
    if action.action_type == "FOLLOW_UP":
        if not state.conversation_history:
            return True  # Nothing happened yet — FOLLOW_UP makes no sense

        agent_turns = [
            e for e in state.conversation_history
            if e.get("speaker") == "agent"
        ]
        prospect_turns = [
            e for e in state.conversation_history
            if e.get("speaker") == "prospect"
        ]

        if not agent_turns:
            return True

        last_agent_turn_num = agent_turns[-1]["turn"]
        last_prospect_turn_num = max(
            (e["turn"] for e in prospect_turns),
            default=0,
        )

        # Violation if prospect already responded AFTER the last agent turn
        return last_prospect_turn_num >= last_agent_turn_num

    return False


def _disqualify_logic(
    state: SalesPathState,
    action: SalesPathAction,
) -> bool:
    """
    R08:
    DISQUALIFY is correct ONLY when:
      - true_budget < close_threshold  AND
      - decision_maker is False
    Violation if prospect is actually closeable OR has a decision maker.
    FIX: Both conditions must hold for a valid disqualification.
    """
    if action.action_type == "DISQUALIFY":
        true_budget = state.hidden_state.get("true_budget", 0.5)
        close_threshold = state.hidden_state.get("close_threshold", 0.5)
        decision_maker = state.prospect_profile.get("decision_maker", True)

        # Valid disqualify requires: low budget AND no decision maker
        valid_disqualify = (true_budget < close_threshold) and (not decision_maker)
        return not valid_disqualify  # Violation if NOT a valid disqualify case

    return False


def _close_requires_demo(
    state: SalesPathState,
    action: SalesPathAction,
) -> bool:
    """
    R09:
    Difficulty 2+ requires OFFER_DEMO before CLOSE.
    """
    if action.action_type == "CLOSE":
        if state.difficulty >= 2:
            return "OFFER_DEMO" not in state.steps_completed
    return False


BUSINESS_RULES = [
    BusinessRule(
        "R01",
        "qualify_before_present",
        "Must QUALIFY before PRESENT",
        _qualify_before_present,
    ),
    BusinessRule(
        "R02",
        "demo_before_negotiate",
        "Must OFFER_DEMO before NEGOTIATE",
        _demo_before_negotiate,
    ),
    BusinessRule(
        "R03",
        "budget_known_to_negotiate",
        "Budget must be known before NEGOTIATE",
        _budget_known_to_negotiate,
    ),
    BusinessRule(
        "R04",
        "discount_after_objections",
        "Discount only after 2 objections handled",
        _discount_after_objections,
    ),
    BusinessRule(
        "R05",
        "no_repeat_action",
        "Cannot repeat same action consecutively",
        _no_repeat_action,
    ),
    BusinessRule(
        "R06",
        "prospect_first",
        "First action must be PROSPECT",
        _prospect_first,
    ),
    BusinessRule(
        "R07",
        "followup_timing",
        "FOLLOW_UP only after prospect silence",
        _followup_timing,
    ),
    BusinessRule(
        "R08",
        "disqualify_logic",
        "DISQUALIFY only when prospect is genuinely unqualified",
        _disqualify_logic,
    ),
    BusinessRule(
        "R09",
        "close_requires_demo",
        "Must OFFER_DEMO before CLOSE (difficulty 2+)",
        _close_requires_demo,
    ),
]


def check_rules(
    state: SalesPathState,
    action: SalesPathAction,
) -> list[str]:
    """
    Returns list of violated rule IDs.
    """

    violated = []

    for rule in BUSINESS_RULES:
        if rule.check(state, action):
            violated.append(rule.rule_id)

    return violated