Spaces:
Paused
Paused
File size: 8,045 Bytes
287f3d3 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 | """Discovery agent / conversation loop tests."""
from __future__ import annotations
import json
import pytest
from agentic_core.orchestrator import OrchestrationError
from agentic_core.schemas import DiscoveryOutput, DiscoveryQuestion
from tests.helpers import discovery_output, build_handler
async def test_vague_idea_asks_for_clarification(provider, make_orchestrator, make_context):
provider.set_handler(build_handler(discovery_status="needs_clarification"))
orchestrator = make_orchestrator()
context = make_context("I want to build a booking platform for football fields.")
output = await orchestrator.discovery_turn(context, context.business_idea)
assert output.status == "needs_clarification"
assert output.questions
assert context.status == "discovery"
assert len(context.transcript) == 2 # user + agent
assert context.transcript[0].role == "user"
assert context.transcript[1].role == "agent"
async def test_complete_idea_reaches_ready(provider, make_orchestrator, make_context):
provider.set_handler(build_handler(discovery_status="ready"))
orchestrator = make_orchestrator()
context = make_context("Food delivery platform with online payment.")
output = await orchestrator.discovery_turn(context, context.business_idea)
assert output.status == "ready"
assert output.questions == []
assert context.status == "ready_for_confirmation"
assert context.target_users == ["Customers", "Restaurants"]
assert context.user_roles == ["Customer", "Restaurant Owner", "Admin"]
async def test_confirmation_requires_ready(provider, make_orchestrator, make_context):
provider.set_handler(build_handler(discovery_status="needs_clarification"))
orchestrator = make_orchestrator()
context = make_context("Something vague.")
await orchestrator.discovery_turn(context, context.business_idea)
with pytest.raises(OrchestrationError):
orchestrator.confirm(context)
provider.set_handler(build_handler(discovery_status="ready"))
context2 = make_context("A clear, complete idea.")
await orchestrator.discovery_turn(context2, context2.business_idea)
orchestrator.confirm(context2)
assert context2.status == "confirmed"
async def test_agent_never_loses_conversation_history(provider, make_orchestrator, make_context):
"""The discovery prompt must include prior turns so the agent never re-asks."""
responses = [
json.dumps(discovery_output("needs_clarification")),
json.dumps(discovery_output("ready")),
]
provider.set_responses(responses)
orchestrator = make_orchestrator()
context = make_context("Booking platform.")
await orchestrator.discovery_turn(context, context.business_idea)
await orchestrator.discovery_turn(context, "Players, field owners and admins.")
second_call_user = provider.calls[1][1]
assert "Booking platform." in second_call_user
assert "Players, field owners and admins." in second_call_user
assert context.transcript[-1].role == "agent"
assert context.status == "ready_for_confirmation"
async def test_contradictory_information_last_answer_wins(provider, make_orchestrator, make_context):
provider.set_handler(
lambda _s, _u: json.dumps(
discovery_output("needs_clarification", ) | {"known_information": {"target_users": ["Players"]}}
)
)
orchestrator = make_orchestrator()
context = make_context("Football field booking.")
await orchestrator.discovery_turn(context, context.business_idea)
assert context.target_users == ["Players"]
provider.set_handler(
lambda _s, _u: json.dumps(
discovery_output("ready") | {"known_information": {"target_users": ["Owners"]}}
)
)
await orchestrator.discovery_turn(context, "Actually owners only.")
assert context.target_users == ["Owners"]
async def test_unnecessary_information_not_forced(provider, make_orchestrator, make_context):
"""not_applicable fields produce no questions."""
payload = discovery_output("needs_clarification")
payload["missing_information"] = [
{"field": "payments", "importance": "not_applicable", "reason": "No payments in this project."}
]
payload["questions"] = []
provider.set_handler(lambda _s, _u: json.dumps(payload))
orchestrator = make_orchestrator()
context = make_context("Internal tool, no payments.")
output = await orchestrator.discovery_turn(context, context.business_idea)
assert output.status == "needs_clarification"
assert output.missing_information[0].importance == "not_applicable"
async def test_known_information_application_is_idempotent(provider, make_orchestrator, make_context):
def handler(_s, _u):
return json.dumps(
discovery_output("ready") | {"known_information": {"core_features": ["Booking", "Payments"]}}
)
provider.set_handler(handler)
orchestrator = make_orchestrator()
context = make_context("Football booking.")
context.core_features = ["Booking", "Payments"]
await orchestrator.discovery_turn(context, context.business_idea)
assert context.core_features == ["Booking", "Payments"]
async def test_discovery_questions_support_options(provider, make_orchestrator, make_context):
"""Questions can carry multiple-choice options; users pick one or type their own."""
payload = discovery_output("needs_clarification")
payload["questions"] = [
{
"id": "q1",
"question": "Which client should v1 ship on?",
"reason": "Defines the build target.",
"options": ["Web app only", "Mobile app only", "Both web and mobile"],
}
]
provider.set_handler(lambda _s, _u: json.dumps(payload))
orchestrator = make_orchestrator()
context = make_context("Booking platform.")
output = await orchestrator.discovery_turn(context, context.business_idea)
question = output.questions[0]
assert isinstance(question, DiscoveryQuestion)
assert question.options == ["Web app only", "Mobile app only", "Both web and mobile"]
assert output.status == "needs_clarification"
async def test_discovery_failure_raises(provider, make_orchestrator, make_context):
async def boom(_s, _u):
raise RuntimeError("provider down")
provider.set_handler(boom)
orchestrator = make_orchestrator()
context = make_context("Football booking.")
with pytest.raises(Exception):
await orchestrator.discovery_turn(context, context.business_idea)
async def test_discovery_retries_transient_failure(provider, make_orchestrator, make_context):
"""A single unparseable reply must not kill discovery; it retries once."""
calls = {"n": 0}
def flaky(_s, _u):
calls["n"] += 1
if calls["n"] <= 2: # first run: generate + repair both fail
return "sorry, no json here"
return json.dumps(discovery_output("ready"))
provider.set_handler(flaky)
orchestrator = make_orchestrator()
context = make_context("Football booking.")
output = await orchestrator.discovery_turn(context, context.business_idea)
assert calls["n"] == 3 # 2 (failed run) + 1 (successful retry run)
assert output.status == "ready"
assert context.status == "ready_for_confirmation"
async def test_discovery_fails_after_two_bad_replies(provider, make_orchestrator, make_context):
calls = {"n": 0}
def always_bad(_s, _u):
calls["n"] += 1
return "still not json"
provider.set_handler(always_bad)
orchestrator = make_orchestrator()
context = make_context("Football booking.")
with pytest.raises(Exception):
await orchestrator.discovery_turn(context, context.business_idea)
assert calls["n"] == 4 # two runs, each generate + repair |