Joseph Doherty
129 lines
3.9 KiB
Python
129 lines
3.9 KiB
Python
"""Tests for the thread-detection service (T55).
|
|
|
|
On scene close, the transcript is classified to detect open threads
|
|
(unresolved arcs, dangling questions, promises made). The service can
|
|
also signal **update** to an existing thread when the scene developed
|
|
it, or **close** when the scene resolved it.
|
|
|
|
These tests cover:
|
|
|
|
* A new thread the scene introduced — action="open" with a fresh title.
|
|
* An update to an existing thread — action="update" with
|
|
``existing_thread_id`` referencing the prior thread.
|
|
* Classifier failure — three bad responses degrade to an empty
|
|
candidates list (graceful degradation, §3.3).
|
|
* Empty transcript short-circuits before any classifier call.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from chat.llm.mock import MockLLMClient
|
|
from chat.services.thread_detection import (
|
|
ThreadCandidate,
|
|
ThreadDetectionResult,
|
|
detect_threads,
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_detects_new_thread_open():
|
|
canned = json.dumps(
|
|
{
|
|
"candidates": [
|
|
{
|
|
"action": "open",
|
|
"title": "Maya's job hunt",
|
|
"summary": "Maya is looking for a new job",
|
|
"existing_thread_id": None,
|
|
}
|
|
]
|
|
}
|
|
)
|
|
mock = MockLLMClient(canned=[canned])
|
|
result = await detect_threads(
|
|
mock,
|
|
classifier_model="x",
|
|
scene_transcript=[
|
|
{"speaker": "Maya", "text": "I need to find a new job soon."},
|
|
{"speaker": "Sam", "text": "What kind of role are you looking for?"},
|
|
],
|
|
open_threads=[],
|
|
)
|
|
assert isinstance(result, ThreadDetectionResult)
|
|
assert len(result.candidates) == 1
|
|
cand = result.candidates[0]
|
|
assert isinstance(cand, ThreadCandidate)
|
|
assert cand.action == "open"
|
|
assert cand.title == "Maya's job hunt"
|
|
assert cand.summary == "Maya is looking for a new job"
|
|
assert cand.existing_thread_id is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_detects_update_to_existing_thread():
|
|
canned = json.dumps(
|
|
{
|
|
"candidates": [
|
|
{
|
|
"action": "update",
|
|
"title": "",
|
|
"summary": "Maya interviewed at Acme today",
|
|
"existing_thread_id": "thr_jobhunt",
|
|
}
|
|
]
|
|
}
|
|
)
|
|
mock = MockLLMClient(canned=[canned])
|
|
result = await detect_threads(
|
|
mock,
|
|
classifier_model="x",
|
|
scene_transcript=[
|
|
{"speaker": "Maya", "text": "I had the Acme interview today."},
|
|
{"speaker": "Sam", "text": "How did it go?"},
|
|
],
|
|
open_threads=[
|
|
{
|
|
"thread_id": "thr_jobhunt",
|
|
"title": "Maya's job hunt",
|
|
"summary": "Maya is looking for a new job",
|
|
}
|
|
],
|
|
)
|
|
assert len(result.candidates) == 1
|
|
cand = result.candidates[0]
|
|
assert cand.action == "update"
|
|
assert cand.existing_thread_id == "thr_jobhunt"
|
|
assert cand.summary == "Maya interviewed at Acme today"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_classifier_failure_returns_empty():
|
|
"""Three malformed classifier responses → empty candidates list."""
|
|
mock = MockLLMClient(canned=["not json", "still not json", "{bad"])
|
|
result = await detect_threads(
|
|
mock,
|
|
classifier_model="x",
|
|
scene_transcript=[
|
|
{"speaker": "Maya", "text": "Anything could happen here."},
|
|
],
|
|
open_threads=[],
|
|
)
|
|
assert result.candidates == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_empty_transcript_short_circuits():
|
|
"""Empty transcript short-circuits — classifier must not be called."""
|
|
mock = MockLLMClient(canned=[])
|
|
result = await detect_threads(
|
|
mock,
|
|
classifier_model="x",
|
|
scene_transcript=[],
|
|
open_threads=[],
|
|
)
|
|
assert result.candidates == []
|