"""Perplexity is a model provider over its Agent API and a web search backend over its Search API, on one API key. The Agent API speaks the OpenAI Responses schema as Perplexity documents it, so the assertions here are on what the daemon sends: the route, the body's own shape (`type` on every input item, no fields Perplexity does not take), and how a tool call and its result are replayed. The model list is the one the Agent API publishes. A key saved through `/auth/keys` is what authenticates both halves, and the Search API's answers become the search's sources. The search order and the saved key are global, so the class is exclusive. """ import ast import json import threading import unittest import urllib.parse import urllib.error from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from harness import ( Albedo, Provider, daemon_resource, error, exclusive, exclusive_class, operation_id, python, text, ) MODEL = "openai/gpt-5.6-sol" KEY = "pplx-fixture-key" AGENT_MODELS = {"data": [{"id": MODEL}, {"id": "perplexity/kimi-k3"}]} # What models.dev knows about the ids the Agent API lists. albedo keeps lookups # there, so this is where the anthropic cap in the tests comes from. MODELS_DEV = { "openai": { "id": "openai", "name": "OpenAI", "models": { "gpt-5.6-sol": { "id": "gpt-5.6-sol", "limit": {"context": 272000, "output": 128000}, } }, }, # models.dev's own namespaces for the same endpoint. Its list is staler than # the Agent API's (a retired id here, a missing one below), so the picker # answering from these instead of the API is exactly what the ordering test # catches. "perplexity": { "id": "perplexity", "name": "Perplexity", "models": { "sonar-retired": { "id": "sonar-retired", "limit": {"context": 128000, "output": 8192}, } }, }, "perplexity-agent": { "id": "perplexity-agent", "name": "Perplexity Agent", "api": "https://api.perplexity.ai/v1", "env": ["PERPLEXITY_API_KEY"], "models": { "sonar-retired": { "id": "sonar-retired", "limit": {"context": 128000, "output": 8192}, } }, }, "anthropic": { "id": "anthropic", "name": "Anthropic", "models": { "claude-sonnet-5": { "id": "claude-sonnet-5", "limit": {"context": 200000, "output": 32000}, "reasoning_options": [ {"type": "effort", "values": ["low", "medium", "high"]} ], } }, }, } RESULTS = [ { "title": "Gleam one", "url": "https://gleam.run/news/one", "snippet": " the first page's text\n", "date": "2026-08-01", "last_updated": "2026-08-02", }, {"title": None, "url": "https://example.com/two", "snippet": "", "date": None}, {"title": "duplicate", "url": "https://gleam.run/news/one", "snippet": "again"}, ] SEARCH = ( "r = await web_search('latest gleam release', limit=2)\n" "print(repr([r.answer, [(s.title, s.url, s.snippet, s.published)" " for s in r.sources]]))\n" ) FAILING = ( "try:\n" " await web_search('latest gleam release')\n" "except Exception as error:\n" " print('raised:', error)\n" ) class FakeSearch: """Answers /search with RESULTS and records what each request sent.""" def __init__(self): self.requests = [] owner = self class Handler(BaseHTTPRequestHandler): def do_POST(self): body = json.loads(self.rfile.read(int(self.headers["Content-Length"]))) owner.requests.append((self.path, self.headers["authorization"], body)) payload = json.dumps({"results": RESULTS, "id": "fixture"}).encode() self.send_response(200) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(payload))) self.end_headers() self.wfile.write(payload) def log_message(self, format, *args): pass self.server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) self.url = f"http://127.0.0.1:{self.server.server_address[1]}" threading.Thread(target=self.server.serve_forever, daemon=True).start() def close(self): self.server.shutdown() self.server.server_close() def scripting(replies): """Answers each request with the next scripted reply.""" def reply(request): return replies.pop(0) if replies else text("done") return reply @exclusive_class class PerplexityTests(unittest.TestCase): def setUp(self): self.search = daemon_resource("search", FakeSearch) self.search.requests.clear() def start(self, replies, *, key=True, prepare=None): provider = Provider( scripting(list(replies)), catalog=AGENT_MODELS, ) self.addCleanup(provider.close) profile = { "extension": "perplexity", "baseUrl": provider.url, "model": MODEL, "protocol": "responses", } if key: profile["apiKey"] = KEY def offline(app): (app.home / "models.json").write_text(json.dumps(MODELS_DEV)) app.write_extensions({"perplexity": {"searchEndpoint": self.search.url}}) if prepare: prepare(app) app = Albedo(provider, providers={"perplexity": profile}, prepare=offline) app.__enter__() self.addCleanup(app.__exit__, None, None, None) return app, provider def put_key(self, app, endpoint="https://api.perplexity.ai/v1"): """Saves a key account the way /login does, and returns its id.""" account = operation_id() with app.api( f"/auth/keys/{account}", { "provider": "perplexity", "endpoint": endpoint, "api_key": KEY, "label": "fixture", }, method="PUT", ) as response: saved = json.load(response) self.assertEqual(saved["kind"], "key") return account def pinned(self, app, account): route = "/settings?group=providers" with app.api(route) as response: revision = response.headers["ETag"] with app.api( route, {"profiles": {app.profile: {"account_id": account}}}, method="PATCH", headers={ "If-Match": revision, "Content-Type": "application/merge-patch+json", }, ) as response: json.load(response) def ask(self, app, prompt="hello"): session = app.session() app.prompt(session, prompt).close() app.idle(session) return session def output(self, app, session): """What the last python cell printed.""" results = [ json.loads(part["value"]) for entry in app.history(session)["items"] if entry["kind"] == "tool_result" and entry["tool"]["name"] == "python" for part in entry["content"] if part["kind"] == "json" and part["field"] == "result" ] return results[-1]["output"] def test_priority_toggle_changes_real_agent_requests_without_changing_model(self): app, provider = self.start( [text("standard"), text("priority"), text("standard again")] ) session = app.session() route = f"/sessions/{session}/fast-mode" for wanted in (False, True, False): with app.api(route) as response: resource = json.load(response) revision = response.headers["ETag"] self.assertEqual(resource["support"]["status"], "supported") self.assertEqual(resource["support"]["mode"], "priority") if resource["fast"] != wanted: with app.api( route, {"state": "toggle"}, method="PATCH", headers={"If-Match": revision}, ) as response: self.assertEqual(json.load(response)["fast"], wanted) app.prompt(session, f"priority requested: {wanted}").close() app.idle(session) sent = provider.requests[-1] self.assertTrue(sent["path"].endswith("/agent")) self.assertEqual(sent["request"]["model"], MODEL) if wanted: self.assertEqual(sent["request"]["service_tier"], "priority") else: self.assertNotIn("service_tier", sent["request"]) self.assertEqual(len(provider.requests), 3) def test_unsupported_priority_model_is_rejected_before_an_agent_request(self): app, provider = self.start([]) settings = "/settings?group=providers" with app.api(settings) as response: revision = response.headers["ETag"] with app.api( settings, {"profiles": {app.profile: {"model": "perplexity/kimi-k3"}}}, method="PATCH", headers={ "If-Match": revision, "Content-Type": "application/merge-patch+json", }, ) as response: json.load(response) session = app.session() route = f"/sessions/{session}/fast-mode" with app.api(route) as response: initial = json.load(response) revision = response.headers["ETag"] self.assertEqual(initial["support"]["status"], "unsupported") with self.assertRaises(urllib.error.HTTPError) as caught: app.api( route, {"state": "on"}, method="PATCH", headers={"If-Match": revision} ) self.assertEqual(caught.exception.code, 400) self.assertEqual(json.load(caught.exception)["code"], "fast_unsupported") with app.api(route) as response: self.assertFalse(json.load(response)["fast"]) self.assertEqual(response.headers["ETag"], revision) self.assertEqual(provider.requests, []) def test_the_agent_api_gets_perplexitys_own_responses_body(self): app, provider = self.start([text("an answer")]) session = self.ask(app, "what is new") [request] = provider.requests self.assertTrue(request["path"].endswith("/agent"), request["path"]) self.assertEqual(request["authorization"], f"Bearer {KEY}") self.assertEqual(request["model"], MODEL) body = request["request"] self.assertTrue(body["stream"]) self.assertFalse(body["store"]) # A session that named no output cap sends none: Perplexity's own # default applies, as it does for every other provider here. self.assertNotIn("max_output_tokens", body) self.assertTrue(body["instructions"]) # Perplexity discriminates its input items by type, and takes none of # the OpenAI fields it documents no support for. self.assertEqual( body["input"], [{"type": "message", "role": "user", "content": "what is new"}], ) for absent in ("include", "parallel_tool_calls", "text", "prompt_cache_key"): self.assertNotIn(absent, body) answers = [ entry for entry in app.history(session)["items"] if entry["kind"] == "assistant" ] self.assertEqual( answers[-1]["content"], [{"kind": "text", "text": "an answer"}] ) def test_an_anthropic_model_gets_the_cap_its_agent_api_requires(self): app, provider = self.start([text("capped")]) # Perplexity answers 400 to an anthropic/* request without a cap, so # that family gets its model's published maximum instead. route = "/settings?group=providers" with app.api(route) as response: revision = response.headers["ETag"] with app.api( route, {"profiles": {app.profile: {"model": "anthropic/claude-sonnet-5"}}}, method="PATCH", headers={ "If-Match": revision, "Content-Type": "application/merge-patch+json", }, ) as response: json.load(response) self.ask(app) body = provider.requests[0]["request"] self.assertEqual(body["model"], "anthropic/claude-sonnet-5") # Exactly what models.dev publishes for that model: a wrong field would # still be a positive number. self.assertEqual(body["max_output_tokens"], 32000) def test_an_unsupported_model_names_the_model_picker(self): # The Agent API answers 400, not 404, for a model its catalog dropped; # the person is told to pick another rather than handed the raw body. app, _provider = self.start( [ error( 400, 'validation failed: model "openai/does-not-exist" is not supported', ) ] ) session = app.session() app.prompt(session, "hi").close() snapshot = app.idle(session, timeout=60) errors = [ line["text"] for line in snapshot["activity"]["lines"] if line["kind"] == "error" ] self.assertTrue( any("choose another in /model" in text for text in errors), errors ) # exclusive: its profile has no key, unlike the class daemon's @exclusive def test_a_saved_key_account_serves_the_agent_api(self): app, provider = self.start([text("keyed")], key=False) # The account's own endpoint is where a pinned key sends its requests. self.pinned(app, self.put_key(app, endpoint=provider.url)) self.ask(app) self.assertEqual(provider.requests[0]["authorization"], f"Bearer {KEY}") def test_a_tool_call_replays_beside_its_result(self): app, provider = self.start([python("print(6 * 7)"), text("it is 42")]) self.ask(app, "multiply") self.assertEqual(len(provider.requests), 2) replay = provider.requests[1]["request"]["input"] call = next(item for item in replay if item.get("type") == "function_call") self.assertEqual(call["name"], "python") self.assertEqual(call["call_id"], "fixture-call-1") result = next( item for item in replay if item.get("type") == "function_call_output" ) self.assertEqual(result["call_id"], "fixture-call-1") self.assertIn("42", result["output"]) def test_the_model_list_is_the_one_the_agent_api_lists(self): app, provider = self.start([]) query = urllib.parse.urlencode( {"provider": "perplexity", "endpoint": provider.url} ) with app.api(f"/models?{query}") as response: listed = [item["id"] for item in json.load(response)["items"]] # The API's own list, not models.dev's `perplexity` / `perplexity-agent` # namespaces, which a later-registered catalog would have answered with. self.assertEqual(listed, [MODEL, "perplexity/kimi-k3"]) self.assertNotIn("sonar-retired", listed) # exclusive: prepares its own web-search settings @exclusive def test_search_answers_with_the_search_apis_sources(self): def prepare(app): (app.home / "models.json").write_text(json.dumps(MODELS_DEV)) app.write_extensions( { "perplexity": {"searchEndpoint": self.search.url}, "web-search": { "order": ["perplexity"], "off": ["codex", "claude", "antigravity", "exa"], }, } ) # The session runs on the harness's own profile: the Search API's key # belongs to the perplexity extension, not to whichever model answers. provider = Provider(scripting([python(FAILING)]), catalog=AGENT_MODELS) self.addCleanup(provider.close) app = Albedo(provider, prepare=prepare) app.__enter__() self.addCleanup(app.__exit__, None, None, None) session = app.session() # No key yet: the search names what is missing instead of reaching out. app.prompt(session, "search without a key").close() app.idle(session) self.assertEqual(self.search.requests, []) self.assertIn("Perplexity needs an API key", self.output(app, session)) self.put_key(app) provider.script = scripting([python(SEARCH)]) app.prompt(session, "search").close() app.idle(session) path, authorization, body = self.search.requests[-1] self.assertEqual(path, "/search") self.assertEqual(authorization, f"Bearer {KEY}") self.assertEqual(body["query"], "latest gleam release") self.assertEqual(body["max_results"], 2) self.assertGreater(body["max_tokens_per_page"], 0) # One source per url, cut to the limit, with the page's own text. self.assertEqual( ast.literal_eval(self.output(app, session).splitlines()[0]), [ "", [ ( "Gleam one", "https://gleam.run/news/one", "the first page's text", "2026-08-01", ), ("https://example.com/two", "https://example.com/two", "", None), ], ], )