WIP: feat: Most efficient deep research system ever made #32

Draft
typosaurus wants to merge 13 commits from typosaurus/31-most-efficient-deep-research-system-ever-made into main
pull from: typosaurus/31-most-efficient-deep-research-system-ever-made
merge into: typosaurus:main
Conversation 0 Commits 13 Files Changed 17 +700
1 changed files with 700 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit be31a2774f - Show all commits

700
tests/test_research_client.py Normal file
View File

@ -0,0 +1,700 @@
# retoor <retoor@molodetz.nl>
import io
import json
import unittest
import urllib.error
from typing import Any
from unittest import mock
from typosaurus_sandbox.research.client import RsearchClient, RsearchError
from typosaurus_sandbox.research.envelopes import (
ChatResponse,
ChatUsage,
DeepReport,
DescribeResponse,
SearchGrade,
SearchResponse,
SearchResult,
)
WEB_RESPONSE: dict[str, Any] = {
"query": "asyncio python",
"source": "duckduckgo",
"count": 3,
"success": True,
"error": None,
"timestamp": "2026-08-07T12:00:00Z",
"results": [
{
"title": "asyncio documentation",
"url": "https://docs.python.org/3/library/asyncio.html",
"description": "Asynchronous I/O event loop.",
"source": "docs.python.org",
"extra": {"rank": 1},
"index": 0,
},
{
"title": "asyncio in Python",
"url": "https://example.com/asyncio",
"description": "Tutorial on asyncio.",
"source": "example.com",
"extra": {"rank": 2},
"index": 1,
},
],
}
AI_MEMORY_RESPONSE: dict[str, Any] = {
"query": "python history",
"source": "ai",
"count": 0,
"success": True,
"error": None,
"results": [],
"ai_response": "From memory: Python was released in 1991 by Guido van Rossum.",
"ai_error": None,
}
AI_PROVIDER_RESPONSE: dict[str, Any] = {
"query": "quantum computing",
"source": "google",
"count": 1,
"success": True,
"error": None,
"results": [
{
"title": "Quantum computing overview",
"url": "https://example.com/quantum",
"description": "Overview of quantum computing.",
"source": "example.com",
"extra": {},
"index": 0,
}
],
"ai_response": "Quantum computing uses qubits. [citation:1]",
"ai_error": None,
}
GRADED_RESPONSE: dict[str, Any] = {
"query": "deep research",
"source": "google",
"count": 1,
"success": True,
"error": None,
"results": [
{
"title": "Deep research systems",
"url": "https://example.com/deep-research",
"description": "Survey of deep research systems.",
"source": "example.com",
"extra": {},
"index": 0,
"grade": {
"overall": 9.2,
"relevance": 8.8,
"depth": 9.0,
"authority": 9.5,
"freshness": 7.0,
"word_count": 1200,
"intent_hits": 4,
},
}
],
}
DEEP_RESPONSE: dict[str, Any] = {
"query": "deep research systems",
"source": "google",
"count": 8,
"success": True,
"error": None,
"results": [
{
"title": "Deep research systems",
"url": "https://example.com/deep-research",
"description": "Survey of deep research systems.",
"source": "example.com",
"extra": {},
"index": 0,
}
],
"deep": {
"query": "deep research systems",
"markdown": "# Deep research\n\nA survey.",
"sources": [
{
"title": "Deep research systems",
"url": "https://example.com/deep-research",
"description": "Survey of deep research systems.",
"source": "example.com",
"extra": {},
"grade": {
"overall": 9.2,
"relevance": 8.8,
"depth": 9.0,
"authority": 9.5,
"freshness": 7.0,
"word_count": 1200,
"intent_hits": 4,
},
}
],
"graded_count": 8,
"total_count": 10,
"model": "gemma-3-12b-it",
"elapsed": 166.96,
"cache_hit": False,
"rounds": 3,
"queries_tried": ["deep research systems", "deep research architecture"],
"error": None,
},
}
IMAGES_RESPONSE: dict[str, Any] = {
"query": "aurora borealis",
"source": "wikimedia",
"count": 2,
"success": True,
"error": None,
"results": [
{
"title": "Aurora borealis over Norway",
"url": "https://commons.wikimedia.org/wiki/File:Aurora.jpg",
"description": "Photograph of the aurora borealis.",
"source": "wikimedia",
"extra": {
"thumbnail": "https://upload.wikimedia.org/thumb.jpg",
"dimensions": {"width": 1920, "height": 1080},
"mime": "image/jpeg",
"license": "CC BY-SA 4.0",
},
"index": 0,
}
],
}
CHAT_RESPONSE: dict[str, Any] = {
"response": "The answer.",
"prompt": "question",
"json_mode": True,
"cached": False,
"error": None,
"usage": {
"prompt_tokens": 120,
"completion_tokens": 80,
"total_tokens": 200,
"cost_usd": 0.0012,
},
}
DESCRIBE_RESPONSE: dict[str, Any] = {
"url": "https://example.com/page",
"description": "Page description",
"elapsed": 1.23,
"timestamp": "2026-08-07T12:00:00Z",
}
SEARCH_EMPTY_OK: dict[str, Any] = {
"query": "q",
"source": "s",
"count": 1,
"success": True,
"error": None,
"results": [],
}
def _recorded_request(fixture: dict[str, Any]) -> tuple[list[tuple[Any, ...]], Any]:
recorded: list[tuple[Any, ...]] = []
def fake(
method: str,
path: str,
params: dict[str, str] | None,
payload: bytes | None,
headers: dict[str, str] | None,
timeout: float | None,
) -> tuple[int, dict[str, Any]]:
recorded.append((method, path, params, payload, headers, timeout))
return 200, fixture
return recorded, fake
def _raising_request(message: str, status_code: int) -> Any:
def fake(
method: str,
path: str,
params: dict[str, str] | None,
payload: bytes | None,
headers: dict[str, str] | None,
timeout: float | None,
) -> tuple[int, dict[str, Any]]:
raise RsearchError(message, status_code)
return fake
class _FakeResponse:
def __init__(self, status: int, body: bytes) -> None:
self.status = status
self._body = body
def __enter__(self) -> "_FakeResponse":
return self
def __exit__(self, *args: object) -> None:
return None
def read(self) -> bytes:
return self._body
class TestSearchResponseParsing(unittest.TestCase):
def test_web_results_parse_into_search_response(self) -> None:
response = SearchResponse.from_dict(WEB_RESPONSE)
self.assertEqual(response.query, "asyncio python")
self.assertEqual(response.source, "duckduckgo")
self.assertEqual(response.count, 3)
self.assertTrue(response.success)
self.assertIsNone(response.error)
self.assertEqual(response.timestamp, "2026-08-07T12:00:00Z")
self.assertEqual(len(response.results), 2)
first = response.results[0]
self.assertIsInstance(first, SearchResult)
self.assertEqual(first.title, "asyncio documentation")
self.assertEqual(first.url, "https://docs.python.org/3/library/asyncio.html")
self.assertEqual(first.description, "Asynchronous I/O event loop.")
self.assertEqual(first.source, "docs.python.org")
self.assertEqual(first.extra, {"rank": 1})
self.assertEqual(first.index, 0)
self.assertIsNone(first.content)
self.assertIsNone(first.grade)
self.assertIsNone(first.query_origin)
self.assertIsNone(response.ai_response)
self.assertIsNone(response.deep)
def test_ai_memory_variant_parses(self) -> None:
response = SearchResponse.from_dict(AI_MEMORY_RESPONSE)
self.assertEqual(response.source, "ai")
self.assertEqual(response.results, [])
self.assertIn("From memory", response.ai_response)
self.assertIsNone(response.ai_error)
def test_ai_provider_variant_parses(self) -> None:
response = SearchResponse.from_dict(AI_PROVIDER_RESPONSE)
self.assertEqual(response.source, "google")
self.assertEqual(len(response.results), 1)
self.assertIn("[citation:1]", response.ai_response)
self.assertIsNone(response.ai_error)
def test_deep_block_parses_into_deep_report(self) -> None:
response = SearchResponse.from_dict(DEEP_RESPONSE)
self.assertIsNotNone(response.deep)
deep = response.deep
self.assertIsInstance(deep, DeepReport)
self.assertEqual(deep.query, "deep research systems")
self.assertEqual(deep.markdown, "# Deep research\n\nA survey.")
self.assertEqual(deep.graded_count, 8)
self.assertEqual(deep.total_count, 10)
self.assertEqual(deep.model, "gemma-3-12b-it")
self.assertEqual(deep.elapsed, 166.96)
self.assertFalse(deep.cache_hit)
self.assertEqual(deep.rounds, 3)
self.assertEqual(deep.queries_tried, ["deep research systems", "deep research architecture"])
self.assertIsNone(deep.error)
self.assertEqual(len(deep.sources), 1)
source = deep.sources[0]
self.assertIsInstance(source, SearchResult)
self.assertEqual(source.url, "https://example.com/deep-research")
self.assertIsInstance(source.grade, SearchGrade)
self.assertEqual(source.grade.overall, 9.2)
def test_images_results_parse_extra_metadata(self) -> None:
response = SearchResponse.from_dict(IMAGES_RESPONSE)
self.assertEqual(response.source, "wikimedia")
result = response.results[0]
self.assertEqual(result.extra["mime"], "image/jpeg")
self.assertEqual(result.extra["dimensions"], {"width": 1920, "height": 1080})
self.assertEqual(result.extra["license"], "CC BY-SA 4.0")
self.assertIn("thumbnail", result.extra)
def test_result_grade_parses_into_search_grade(self) -> None:
response = SearchResponse.from_dict(GRADED_RESPONSE)
grade = response.results[0].grade
self.assertIsInstance(grade, SearchGrade)
self.assertEqual(grade.overall, 9.2)
self.assertEqual(grade.relevance, 8.8)
self.assertEqual(grade.depth, 9.0)
self.assertEqual(grade.authority, 9.5)
self.assertEqual(grade.freshness, 7.0)
self.assertEqual(grade.word_count, 1200)
self.assertEqual(grade.intent_hits, 4)
def test_sparse_body_parses_with_defaults(self) -> None:
response = SearchResponse.from_dict({"query": "x", "success": True})
self.assertEqual(response.source, "")
self.assertEqual(response.count, 0)
self.assertEqual(response.results, [])
self.assertIsNone(response.error)
self.assertIsNone(response.deep)
self.assertIsNone(response.ai_response)
class TestSearchRequestConstruction(unittest.IsolatedAsyncioTestCase):
async def test_search_forwards_all_parameters(self) -> None:
client = RsearchClient()
recorded, fake = _recorded_request(SEARCH_EMPTY_OK)
client._request = fake
await client.search(
"query text",
source="google",
count=7,
content=True,
type="images",
deep=True,
ai=True,
cache=False,
)
method, path, params, payload, headers, timeout = recorded[0]
self.assertEqual(method, "GET")
self.assertEqual(path, "/search")
self.assertEqual(
params,
{
"query": "query text",
"source": "google",
"count": "7",
"content": "true",
"type": "images",
"deep": "true",
"ai": "true",
"cache": "false",
},
)
self.assertIsNone(payload)
self.assertIsNone(headers)
self.assertEqual(timeout, 180.0)
async def test_search_without_deep_uses_request_timeout(self) -> None:
client = RsearchClient()
recorded, fake = _recorded_request(SEARCH_EMPTY_OK)
client._request = fake
await client.search("q")
self.assertEqual(recorded[0][5], 30.0)
async def test_count_none_omits_count_parameter(self) -> None:
client = RsearchClient()
recorded, fake = _recorded_request(SEARCH_EMPTY_OK)
client._request = fake
await client.search("q")
self.assertNotIn("count", recorded[0][2])
async def test_count_zero_forwarded_and_server_clamp_parsed(self) -> None:
client = RsearchClient()
recorded, fake = _recorded_request(
{"query": "q", "source": "s", "count": 1, "success": True, "error": None, "results": []}
)
client._request = fake
response = await client.search("q", count=0)
self.assertEqual(recorded[0][2]["count"], "0")
self.assertEqual(response.count, 1)
async def test_count_above_limit_forwarded_and_server_clamp_parsed(self) -> None:
client = RsearchClient()
recorded, fake = _recorded_request(
{"query": "q", "source": "s", "count": 10, "success": True, "error": None, "results": []}
)
client._request = fake
response = await client.search("q", count=25)
self.assertEqual(recorded[0][2]["count"], "25")
self.assertEqual(response.count, 10)
async def test_invalid_count_forwarded_and_server_clamp_parsed(self) -> None:
client = RsearchClient()
recorded, fake = _recorded_request(
{"query": "q", "source": "s", "count": 10, "success": True, "error": None, "results": []}
)
client._request = fake
response = await client.search("q", count="not-a-number")
self.assertEqual(recorded[0][2]["count"], "not-a-number")
self.assertEqual(response.count, 10)
async def test_search_cache_hit_skips_second_request(self) -> None:
client = RsearchClient()
recorded, fake = _recorded_request(SEARCH_EMPTY_OK)
client._request = fake
await client.search("cached query")
await client.search("cached query")
self.assertEqual(len(recorded), 1)
async def test_search_cache_disabled_repeats_request(self) -> None:
client = RsearchClient()
recorded, fake = _recorded_request(SEARCH_EMPTY_OK)
client._request = fake
await client.search("uncached query", cache=False)
await client.search("uncached query", cache=False)
self.assertEqual(len(recorded), 2)
async def test_search_with_content_populates_content_cache(self) -> None:
client = RsearchClient()
fixture = {
"query": "q",
"source": "s",
"count": 1,
"success": True,
"error": None,
"results": [
{
"title": "t",
"url": "https://example.com/a",
"description": "d",
"source": "s",
"extra": {},
"content": "full page text",
}
],
}
recorded, fake = _recorded_request(fixture)
client._request = fake
await client.search("q", content=True)
self.assertEqual(len(recorded), 1)
self.assertEqual(client.get_cached_content("https://example.com/a"), "full page text")
async def test_search_error_in_body_surfaces_rsearch_error(self) -> None:
client = RsearchClient()
client._request = _raising_request("Empty query", 400)
with self.assertRaises(RsearchError) as ctx:
await client.search("")
self.assertEqual(ctx.exception.status_code, 400)
self.assertEqual(str(ctx.exception), "Empty query")
class TestChatResponseParsing(unittest.IsolatedAsyncioTestCase):
async def test_chat_response_parses_envelope(self) -> None:
client = RsearchClient()
recorded, fake = _recorded_request(CHAT_RESPONSE)
client._request = fake
response = await client.chat("question", json_mode=True)
method, path, params, payload, headers, timeout = recorded[0]
self.assertEqual(method, "POST")
self.assertEqual(path, "/chat")
self.assertEqual(json.loads(payload), {"prompt": "question", "json": True})
self.assertEqual(headers, {"Content-Type": "application/json"})
self.assertIsNone(params)
self.assertIsNone(timeout)
self.assertIsInstance(response, ChatResponse)
self.assertEqual(response.response, "The answer.")
self.assertEqual(response.prompt, "question")
self.assertTrue(response.json_mode)
self.assertFalse(response.cached)
self.assertIsNone(response.error)
self.assertIsInstance(response.usage, ChatUsage)
self.assertEqual(response.usage.prompt_tokens, 120)
self.assertEqual(response.usage.completion_tokens, 80)
self.assertEqual(response.usage.total_tokens, 200)
self.assertEqual(response.usage.cost_usd, 0.0012)
async def test_chat_request_accepts_system_and_disables_cache(self) -> None:
client = RsearchClient()
recorded, fake = _recorded_request(CHAT_RESPONSE)
client._request = fake
await client.chat("q", system="sys", cache=False)
body = json.loads(recorded[0][3])
self.assertEqual(body, {"prompt": "q", "system": "sys", "cache": False})
async def test_chat_error_raises_mapped_rsearch_error(self) -> None:
client = RsearchClient()
client._request = _raising_request("No prompt provided", 400)
with self.assertRaises(RsearchError) as ctx:
await client.chat("")
self.assertEqual(ctx.exception.status_code, 400)
self.assertEqual(str(ctx.exception), "No prompt provided")
class TestDescribeResponseParsing(unittest.IsolatedAsyncioTestCase):
async def test_describe_get_parses_envelope(self) -> None:
client = RsearchClient()
recorded, fake = _recorded_request(DESCRIBE_RESPONSE)
client._request = fake
response = await client.describe("https://example.com/page")
method, path, params, payload, headers, timeout = recorded[0]
self.assertEqual(method, "GET")
self.assertEqual(path, "/describe")
self.assertEqual(params, {"url": "https://example.com/page"})
self.assertIsNone(payload)
self.assertIsNone(headers)
self.assertIsNone(timeout)
self.assertIsInstance(response, DescribeResponse)
self.assertEqual(response.description, "Page description")
self.assertEqual(response.url, "https://example.com/page")
self.assertEqual(response.elapsed, 1.23)
self.assertEqual(response.timestamp, "2026-08-07T12:00:00Z")
self.assertIsNone(response.mime_type)
self.assertIsNone(response.size)
self.assertTrue(response.success)
async def test_describe_get_uses_cache(self) -> None:
client = RsearchClient()
recorded, fake = _recorded_request(DESCRIBE_RESPONSE)
client._request = fake
await client.describe("https://example.com/page")
await client.describe("https://example.com/page")
self.assertEqual(len(recorded), 1)
async def test_describe_error_raises_mapped_rsearch_error(self) -> None:
client = RsearchClient()
client._request = _raising_request("No url provided", 400)
with self.assertRaises(RsearchError) as ctx:
await client.describe("")
self.assertEqual(ctx.exception.status_code, 400)
self.assertEqual(str(ctx.exception), "No url provided")
async def test_describe_raw_posts_bytes_with_content_type(self) -> None:
client = RsearchClient()
recorded, fake = _recorded_request(DESCRIBE_RESPONSE)
client._request = fake
image = b"\x89PNG\r\n\x1a\npayload"
await client.describe_raw(image, mime_type="image/png")
method, path, params, payload, headers, timeout = recorded[0]
self.assertEqual(method, "POST")
self.assertEqual(path, "/describe")
self.assertEqual(payload, image)
self.assertEqual(headers, {"Content-Type": "image/png"})
async def test_describe_upload_builds_multipart_body(self) -> None:
client = RsearchClient()
recorded, fake = _recorded_request(DESCRIBE_RESPONSE)
client._request = fake
image = b"\x89PNGpayload"
await client.describe_upload(image, filename="photo.png", mime_type="image/png")
method, path, params, payload, headers, timeout = recorded[0]
self.assertEqual(method, "POST")
self.assertEqual(path, "/describe")
self.assertIn(b'name="file"; filename="photo.png"', payload)
self.assertIn(b"Content-Type: image/png", payload)
self.assertIn(image, payload)
self.assertIn("multipart/form-data; boundary=", headers["Content-Type"])
async def test_describe_raw_reuses_cache_by_hash(self) -> None:
client = RsearchClient()
recorded, fake = _recorded_request(DESCRIBE_RESPONSE)
client._request = fake
image = b"\x89PNGpayload"
await client.describe_raw(image, mime_type="image/png")
await client.describe_raw(image, mime_type="image/png")
self.assertEqual(len(recorded), 1)
class TestErrorInBodyHandling(unittest.TestCase):
def test_empty_query_error_in_body_maps_to_rsearch_error(self) -> None:
client = RsearchClient()
body = b'{"success": false, "error": "Empty query"}'
error = urllib.error.HTTPError(
"https://rsearch.app.molodetz.nl/search", 400, "Bad Request", {}, io.BytesIO(body)
)
with mock.patch("urllib.request.urlopen", side_effect=error):
with self.assertRaises(RsearchError) as ctx:
client._request("GET", "/search", {"query": ""}, None, None, 30.0)
self.assertEqual(ctx.exception.status_code, 400)
self.assertEqual(str(ctx.exception), "Empty query")
def test_providers_exhausted_503_maps_to_rsearch_error(self) -> None:
client = RsearchClient()
body = b'{"success": false, "error": "All providers are exhausted, please try again later"}'
error = urllib.error.HTTPError(
"https://rsearch.app.molodetz.nl/search", 503, "Service Unavailable", {}, io.BytesIO(body)
)
with mock.patch("urllib.request.urlopen", side_effect=error):
with self.assertRaises(RsearchError) as ctx:
client._request("GET", "/search", {"query": "q"}, None, None, 30.0)
self.assertEqual(ctx.exception.status_code, 503)
self.assertEqual(str(ctx.exception), "All providers are exhausted, please try again later")
def test_success_false_body_with_http_200_raises(self) -> None:
client = RsearchClient()
fake = _FakeResponse(200, b'{"success": false, "error": "Empty query"}')
with mock.patch("urllib.request.urlopen", return_value=fake):
with self.assertRaises(RsearchError) as ctx:
client._request("GET", "/search", {"query": ""}, None, None, 30.0)
self.assertEqual(ctx.exception.status_code, 200)
self.assertEqual(str(ctx.exception), "Empty query")
def test_detail_field_falls_back_for_error_message(self) -> None:
client = RsearchClient()
body = b'{"detail": "No url provided"}'
error = urllib.error.HTTPError(
"https://rsearch.app.molodetz.nl/describe", 400, "Bad Request", {}, io.BytesIO(body)
)
with mock.patch("urllib.request.urlopen", side_effect=error):
with self.assertRaises(RsearchError) as ctx:
client._request("GET", "/describe", {"url": "x"}, None, None, 30.0)
self.assertEqual(ctx.exception.status_code, 400)
self.assertEqual(str(ctx.exception), "No url provided")
def test_title_field_falls_back_for_error_message(self) -> None:
client = RsearchClient()
body = b'{"title": "Provider error"}'
error = urllib.error.HTTPError(
"https://rsearch.app.molodetz.nl/search", 502, "Bad Gateway", {}, io.BytesIO(body)
)
with mock.patch("urllib.request.urlopen", side_effect=error):
with self.assertRaises(RsearchError) as ctx:
client._request("GET", "/search", {"query": "q"}, None, None, 30.0)
self.assertEqual(ctx.exception.status_code, 502)
self.assertEqual(str(ctx.exception), "Provider error")
def test_empty_body_raises_rsearch_error(self) -> None:
client = RsearchClient()
fake = _FakeResponse(200, b"")
with mock.patch("urllib.request.urlopen", return_value=fake):
with self.assertRaises(RsearchError) as ctx:
client._request("GET", "/search", {"query": "q"}, None, None, 30.0)
self.assertEqual(ctx.exception.status_code, 200)
self.assertIn("empty response", str(ctx.exception))
def test_invalid_json_body_raises_rsearch_error(self) -> None:
client = RsearchClient()
fake = _FakeResponse(200, b"<html>not json</html>")
with mock.patch("urllib.request.urlopen", return_value=fake):
with self.assertRaises(RsearchError) as ctx:
client._request("GET", "/search", {"query": "q"}, None, None, 30.0)
self.assertEqual(ctx.exception.status_code, 200)
self.assertIn("invalid JSON", str(ctx.exception))
def test_non_dict_body_raises_rsearch_error(self) -> None:
client = RsearchClient()
fake = _FakeResponse(200, b'["not", "a", "dict"]')
with mock.patch("urllib.request.urlopen", return_value=fake):
with self.assertRaises(RsearchError) as ctx:
client._request("GET", "/search", {"query": "q"}, None, None, 30.0)
self.assertEqual(ctx.exception.status_code, 200)
def test_connection_failure_raises_rsearch_error(self) -> None:
client = RsearchClient()
with mock.patch("urllib.request.urlopen", side_effect=urllib.error.URLError("connection refused")):
with self.assertRaises(RsearchError) as ctx:
client._request("GET", "/search", {"query": "q"}, None, None, 30.0)
self.assertIn("connection failure", str(ctx.exception))
def test_successful_request_returns_status_and_body(self) -> None:
client = RsearchClient()
fake = _FakeResponse(200, b'{"success": true, "query": "q", "count": 1, "results": []}')
with mock.patch("urllib.request.urlopen", return_value=fake) as urlopen:
status, data = client._request("GET", "/search", {"query": "q"}, None, None, 30.0)
self.assertEqual(status, 200)
self.assertEqual(data, {"success": True, "query": "q", "count": 1, "results": []})
request = urlopen.call_args[0][0]
self.assertEqual(request.get_method(), "GET")
self.assertEqual(request.get_full_url(), "https://rsearch.app.molodetz.nl/search?query=q")
if __name__ == "__main__":
unittest.main()