3c22d88bcd
- Demo accounts: super_admin-only provisioning into isolated DEMO_ORG_ID tenant, 30-day UTC trial on first login, revocable, one-time credential delivery via optional SES/webhook (never persisted). Adds boto3 dependency. - Analytics/report/export/privacy: shared bounded scan budget across users/groups/ sessions, tenant-consistent session/user/group joins, scalar-only CSV export (no nested persisted-value stringification). - Ownership/tenant isolation: canonical owner-tenant predicate for list/read/chat; client sees is_owned only, never owner_user_id. - Lifecycle/races: status transition validation, analyzing is an in-progress gate (no duplicate reanalysis), structured-ready publication, stale-variant revalidation. - Auth/setup/consent/JWT/OAuth/config: fail-closed consent, bounded JWT lifetime, provider-subject atomic OAuth identity, repeated-secret rejection, strict Persona trait validation. - Chat/session/privacy: pre-seller opener redaction, corrupt-session recovery, role-aware completed-chat dashboard routing. - Frontend: Training→product→personas→practice flow, demo/role/demo guards, is_owned-based ownership display, 320×568 and 500×768 responsive E2E. - 8 independent exact-five-key review scopes passed; backend 509, frontend 26, production build 1775 modules, isolated E2E 15.
360 lines
11 KiB
Python
360 lines
11 KiB
Python
"""Final-judge invariants for automatic and manual finish paths."""
|
|
from __future__ import annotations
|
|
|
|
from copy import deepcopy
|
|
|
|
import pytest
|
|
|
|
from app.api import chat_routes
|
|
from app.llm import LLMError
|
|
|
|
|
|
class FakeSessions:
|
|
def __init__(self, session):
|
|
self.row = deepcopy(session)
|
|
self.updates = []
|
|
|
|
def update(self, sid, **fields):
|
|
assert sid == self.row["id"]
|
|
self.updates.append(fields)
|
|
self.row.update(fields)
|
|
return deepcopy(self.row)
|
|
|
|
|
|
class StubSim:
|
|
def __init__(self, verdict=None, error=False):
|
|
self.verdict = verdict or {
|
|
"outcome": "won",
|
|
"score": 88,
|
|
"pain": "qualified pain",
|
|
"why": "clear fit and acceptance",
|
|
"failurePoints": [],
|
|
"coaching": ["Keep the close concise"],
|
|
"painProgress": {"qualified pain": 100},
|
|
}
|
|
self.error = error
|
|
self.judge_calls = 0
|
|
self.public_debrief_calls = 0
|
|
self.public_verdict: dict | None = None
|
|
self.public_error = False
|
|
|
|
def judge(self, **kwargs):
|
|
self.judge_calls += 1
|
|
if self.error:
|
|
raise LLMError("provider unavailable")
|
|
return deepcopy(self.verdict)
|
|
|
|
def public_debrief(self, **kwargs):
|
|
self.public_debrief_calls += 1
|
|
if self.public_error:
|
|
raise LLMError("public debrief unavailable")
|
|
source = self.public_verdict if self.public_verdict is not None else self.verdict
|
|
return {
|
|
key: deepcopy(source.get(key))
|
|
for key in ("why", "failurePoints", "coaching")
|
|
if key in source
|
|
}
|
|
|
|
|
|
def _session():
|
|
return {
|
|
"id": "session-1",
|
|
"org_id": "org-1",
|
|
"user_id": "user-1",
|
|
"group_id": "group-1",
|
|
"persona_id": "persona-1",
|
|
"persona_name": "Customer",
|
|
"mode": "trainee",
|
|
"status": "active",
|
|
"outcome": None,
|
|
"messages": [{"role": "seller", "text": "สวัสดี"}],
|
|
"internal": {"turns": 2, "score": 61},
|
|
"debrief": None,
|
|
"locale": "th",
|
|
}
|
|
|
|
|
|
def _run(monkeypatch, *, verdict=None, error=False, session=None):
|
|
sessions = FakeSessions(session or _session())
|
|
stores = {"sessions": sessions}
|
|
sim = StubSim(verdict=verdict, error=error)
|
|
monkeypatch.setattr(chat_routes, "_sim", lambda group, persona: sim)
|
|
updated, debrief = chat_routes._finalize_session(
|
|
stores,
|
|
sessions.row,
|
|
{"sales_kit": {}},
|
|
{"name": "Customer", "pains": [{"description": "pain"}]},
|
|
)
|
|
return sessions, sim, updated, debrief
|
|
|
|
|
|
def test_final_judge_supplies_score_and_debrief_for_automatic_buy(monkeypatch):
|
|
sessions, sim, updated, debrief = _run(monkeypatch)
|
|
|
|
assert sim.judge_calls == 1
|
|
assert sim.public_debrief_calls == 1
|
|
assert updated["status"] == "finished"
|
|
assert updated["outcome"] == "won"
|
|
assert debrief["score"] == 88
|
|
assert debrief["coaching"] == ["Keep the close concise"]
|
|
# IP protection: pain is hidden from the debrief (prose + raw list).
|
|
assert "pain" not in debrief
|
|
assert "painProgress" not in debrief
|
|
assert "pains" not in debrief["revealed_persona"]
|
|
assert sessions.updates[0]["messages"]
|
|
|
|
|
|
def test_public_debrief_does_not_use_hidden_judge_prose(monkeypatch):
|
|
sessions = FakeSessions(_session())
|
|
sim = StubSim(
|
|
verdict={
|
|
"outcome": "won",
|
|
"score": 88,
|
|
"why": "secret opener and hidden budget were exposed",
|
|
"failurePoints": ["secret pain and secret objection"],
|
|
"coaching": ["secret formula: use the hidden lever"],
|
|
}
|
|
)
|
|
sim.public_verdict = {
|
|
"why": "The close was concise and direct",
|
|
"failurePoints": [],
|
|
"coaching": ["Keep the close concise"],
|
|
}
|
|
monkeypatch.setattr(chat_routes, "_sim", lambda group, persona: sim)
|
|
|
|
_updated, debrief = chat_routes._finalize_session(
|
|
{"sessions": sessions},
|
|
sessions.row,
|
|
{"sales_kit": {}},
|
|
{
|
|
"name": "Customer",
|
|
"opener": "secret opener",
|
|
"budget": "secret budget",
|
|
"pains": [{"description": "secret pain"}],
|
|
"negotiation_levers": ["secret lever"],
|
|
},
|
|
)
|
|
|
|
assert debrief["why"] == "The close was concise and direct"
|
|
assert debrief["coaching"] == ["Keep the close concise"]
|
|
assert "secret opener" not in str(debrief)
|
|
assert "secret budget" not in str(debrief)
|
|
assert "secret pain" not in str(debrief)
|
|
assert "secret formula" not in str(debrief)
|
|
|
|
|
|
def test_public_debrief_failure_never_falls_back_to_hidden_judge_prose(monkeypatch):
|
|
sessions = FakeSessions(_session())
|
|
sim = StubSim(
|
|
verdict={
|
|
"outcome": "lost",
|
|
"score": 10,
|
|
"why": "secret hidden persona explanation",
|
|
"failurePoints": ["secret pain"],
|
|
"coaching": ["secret coaching formula"],
|
|
}
|
|
)
|
|
sim.public_error = True
|
|
monkeypatch.setattr(chat_routes, "_sim", lambda group, persona: sim)
|
|
|
|
_updated, debrief = chat_routes._finalize_session(
|
|
{"sessions": sessions},
|
|
sessions.row,
|
|
{"sales_kit": {}},
|
|
{"name": "Customer", "pains": [{"description": "secret pain"}]},
|
|
)
|
|
|
|
assert debrief["why"] == ""
|
|
assert debrief["failurePoints"] == []
|
|
assert debrief["coaching"] == []
|
|
assert "secret hidden persona explanation" not in str(debrief)
|
|
assert "secret pain" not in str(debrief)
|
|
|
|
|
|
def test_debrief_never_returns_secret_persona_fields(monkeypatch):
|
|
sessions = FakeSessions(_session())
|
|
sim = StubSim()
|
|
monkeypatch.setattr(chat_routes, "_sim", lambda group, persona: sim)
|
|
persona = {
|
|
"name": "Customer",
|
|
"tier": "B",
|
|
"initiation_mode": "customer",
|
|
"channel": "line",
|
|
"profession": "Consultant",
|
|
"age_group": "30s",
|
|
"location": "Bangkok",
|
|
"product_context": "Sales software",
|
|
"pains": [{"description": "hidden pain"}],
|
|
"objections": ["hidden objection"],
|
|
"negotiation_levers": ["hidden lever"],
|
|
"opener": "secret opener",
|
|
"rootCause": "secret root cause",
|
|
"resolutionConditions": ["secret condition"],
|
|
"tolerance": 1,
|
|
"income": "secret income",
|
|
"lifestyle": "secret lifestyle",
|
|
"personality": "secret personality",
|
|
"communication_style": "secret communication style",
|
|
"budget": "secret budget",
|
|
"decision_timeline": "secret timeline",
|
|
"goal": "secret goal",
|
|
"background": "secret background",
|
|
}
|
|
|
|
_updated, debrief = chat_routes._finalize_session(
|
|
{"sessions": sessions},
|
|
sessions.row,
|
|
{"sales_kit": {}},
|
|
persona,
|
|
)
|
|
|
|
revealed = debrief["revealed_persona"]
|
|
assert set(revealed) == {
|
|
"name",
|
|
"tier",
|
|
"initiation_mode",
|
|
"channel",
|
|
"profession",
|
|
"age_group",
|
|
"location",
|
|
"product_context",
|
|
}
|
|
assert not {
|
|
"pains",
|
|
"objections",
|
|
"negotiation_levers",
|
|
"opener",
|
|
"rootCause",
|
|
"resolutionConditions",
|
|
"tolerance",
|
|
"income",
|
|
"lifestyle",
|
|
"personality",
|
|
"communication_style",
|
|
"budget",
|
|
"decision_timeline",
|
|
"goal",
|
|
"background",
|
|
}.intersection(revealed)
|
|
|
|
|
|
def test_final_judge_supplies_lost_outcome_for_automatic_walk(monkeypatch):
|
|
sessions, sim, updated, debrief = _run(
|
|
monkeypatch,
|
|
verdict={
|
|
"outcome": "lost",
|
|
"score": 17,
|
|
"pain": "unresolved",
|
|
"why": "seller missed the need",
|
|
"failurePoints": ["no discovery"],
|
|
"coaching": ["Ask a deeper question"],
|
|
"painProgress": {"unresolved": 20},
|
|
},
|
|
)
|
|
|
|
assert sim.judge_calls == 1
|
|
assert updated["outcome"] == "lost"
|
|
assert debrief["score"] == 17
|
|
assert debrief["failurePoints"] == ["no discovery"]
|
|
|
|
|
|
@pytest.mark.parametrize("score", [float("inf"), float("-inf")])
|
|
def test_nonfinite_final_judge_score_fails_closed(monkeypatch, score):
|
|
_sessions, _sim, _updated, debrief = _run(
|
|
monkeypatch,
|
|
verdict={
|
|
"outcome": "won",
|
|
"score": score,
|
|
"why": "accepted",
|
|
"failurePoints": [],
|
|
"coaching": [],
|
|
},
|
|
)
|
|
|
|
assert debrief["score"] == 0
|
|
|
|
|
|
def test_manual_finish_uses_the_same_helper_and_score(monkeypatch):
|
|
sessions, sim, updated, debrief = _run(
|
|
monkeypatch,
|
|
verdict={
|
|
"outcome": "won",
|
|
"score": 73,
|
|
"pain": "fit",
|
|
"why": "accepted",
|
|
"failurePoints": [],
|
|
"coaching": [],
|
|
"painProgress": {"fit": 100},
|
|
},
|
|
)
|
|
|
|
assert sim.judge_calls == 1
|
|
assert updated["outcome"] == "won"
|
|
assert debrief["score"] == 73
|
|
|
|
|
|
def test_judge_failure_persists_transcript_but_keeps_session_retryable(monkeypatch):
|
|
session = _session()
|
|
sessions = FakeSessions(session)
|
|
sim = StubSim(error=True)
|
|
monkeypatch.setattr(chat_routes, "_sim", lambda group, persona: sim)
|
|
|
|
with pytest.raises(LLMError):
|
|
chat_routes._finalize_session(
|
|
{"sessions": sessions},
|
|
sessions.row,
|
|
{"sales_kit": {}},
|
|
{"name": "Customer", "pains": []},
|
|
)
|
|
|
|
assert sim.judge_calls == 1
|
|
assert sessions.row["status"] == "active"
|
|
assert sessions.row["outcome"] is None
|
|
assert sessions.updates[0]["messages"] == session["messages"]
|
|
|
|
|
|
def test_malformed_final_judge_result_keeps_session_retryable(monkeypatch):
|
|
sessions = FakeSessions(_session())
|
|
sim = StubSim()
|
|
monkeypatch.setattr(sim, "judge", lambda **_kwargs: ["not", "a", "verdict"])
|
|
monkeypatch.setattr(chat_routes, "_sim", lambda group, persona: sim)
|
|
|
|
with pytest.raises(LLMError):
|
|
chat_routes._finalize_session(
|
|
{"sessions": sessions},
|
|
sessions.row,
|
|
{"sales_kit": {}},
|
|
{"name": "Customer", "pains": []},
|
|
)
|
|
|
|
assert sessions.row["status"] == "active"
|
|
assert sessions.row["outcome"] is None
|
|
|
|
|
|
def test_finalization_is_idempotent_without_a_second_judge(monkeypatch):
|
|
session = _session()
|
|
session.update(
|
|
status="finished",
|
|
outcome="won",
|
|
debrief={
|
|
"outcome": "won",
|
|
"score": 91,
|
|
"pain": "fit",
|
|
"why": "closed",
|
|
"failurePoints": [],
|
|
"coaching": [],
|
|
"painProgress": {},
|
|
},
|
|
)
|
|
sessions = FakeSessions(session)
|
|
sim = StubSim()
|
|
monkeypatch.setattr(chat_routes, "_sim", lambda group, persona: sim)
|
|
|
|
updated, debrief = chat_routes._finalize_session(
|
|
{"sessions": sessions}, sessions.row, {}, {"name": "Customer"}
|
|
)
|
|
|
|
assert sim.judge_calls == 0
|
|
assert updated["status"] == "finished"
|
|
assert debrief["score"] == 91
|