from __future__ import annotations import asyncio import copy import json import sys import tempfile import unittest from pathlib import Path from unittest.mock import patch import httpx ROOT = Path(__file__).resolve().parents[2] sys.path.insert(0, str(ROOT / "backend")) from app.services.visual_review import VisualReviewError, review_candidate_batch, review_checkpoint # noqa: E402 from app.settings import ProviderConfig, ProviderModel, Settings # noqa: E402 BACKEND = ROOT / "backend" def settings(root: Path) -> Settings: provider = ProviderConfig( "review", "Review", "https://review.example.invalid/v1", "test-key", (ProviderModel("review-vision", vision=True),), ) return Settings( task_root=root / "tasks", conversation_root=root / "conversations", library_root=BACKEND / "cdsl_library", engine_root=BACKEND / "engine" / "cdsl_engine", llm_base_url=provider.base_url, llm_api_key=provider.api_key, llm_model="review-vision", llm_timeout_s=1, default_provider_id="review", providers=(provider,), review_provider_id="review", review_model_id="review-vision", ) class _Response: def __init__(self, status_code: int, body: dict[str, object] | None = None, text: str = "") -> None: self.status_code = status_code self._body = body or {} self.text = text def json(self) -> dict[str, object]: return self._body class _Client: def __init__(self, responses: list[_Response]) -> None: self.responses = responses self.requests: list[dict[str, object]] = [] async def __aenter__(self) -> "_Client": return self async def __aexit__(self, *_: object) -> None: return None async def post(self, _url: str, *, headers: dict[str, str], json: dict[str, object]) -> _Response: self.requests.append(copy.deepcopy(json)) return self.responses.pop(0) def _valid_tool_response() -> _Response: arguments = { "verdict": "pass", "confidence": 0.96, "affected_node_ids": [], "requirement_ids": [], "evidence": ["Canonical top and isometric views show the expected silhouette."], } return _Response(200, { "choices": [{"message": {"tool_calls": [{"function": { "name": "review_rendered_checkpoint", "arguments": json.dumps(arguments), }}]}}], }) def _candidate_response(*, verdict: str = "accept", batch_goal_status: str = "achieved") -> _Response: arguments = { "verdict": verdict, "confidence": 0.96, "batch_goal_status": batch_goal_status, "coverage": [{ "item": "one connected plate", "status": "complete", "evidence": "The isometric view shows one continuous rectangular solid.", }], "evidence": ["The staged plate achieves the batch goal."], } return _Response(200, { "choices": [{"message": {"tool_calls": [{"function": { "name": "review_candidate_batch", "arguments": json.dumps(arguments), }}]}}], }) class VisualReviewCompatibilityTests(unittest.TestCase): def _review(self, root: Path, client: _Client, *, source_requirements: str = "") -> dict[str, object]: image = root / "iso.png" image.write_bytes(b"png") manifest = { "renderer": "test", "source": "test", "views": [{"id": "isometric", "path": str(image), "camera": {}}], } with patch("app.services.visual_review.httpx.AsyncClient", return_value=client): return asyncio.run(review_checkpoint( settings(root), manifest=manifest, requirements="# Frozen requirements", source_requirements=source_requirements, deterministic_report={"health": {"valid": True}}, )) def test_checkpoint_review_receives_source_and_frozen_requirements(self) -> None: with tempfile.TemporaryDirectory() as temporary: client = _Client([_valid_tool_response()]) self._review(Path(temporary), client, source_requirements="Create a plate with a mounting hole.") review_input = json.loads(client.requests[0]["messages"][1]["content"][0]["text"]) self.assertEqual(review_input["source_requirements"], "Create a plate with a mounting hole.") self.assertEqual(review_input["requirements"], "# Frozen requirements") self.assertIn("may never remove, replace, or weaken", review_input["instruction"]) def test_retries_without_forced_tool_choice_only_for_thinking_rejection(self) -> None: with tempfile.TemporaryDirectory() as temporary: client = _Client([ _Response(400, text="Thinking mode does not support this tool_choice"), _valid_tool_response(), ]) result = self._review(Path(temporary), client) self.assertEqual(result["verdict"], "pass") self.assertEqual(len(client.requests), 2) self.assertIn("tool_choice", client.requests[0]) self.assertNotIn("tool_choice", client.requests[1]) self.assertIn("tools", client.requests[1]) def test_compatibility_retry_still_rejects_prose_or_missing_tool_call(self) -> None: with tempfile.TemporaryDirectory() as temporary: client = _Client([ _Response(400, text="thinking mode does not support this tool_choice"), _Response(200, {"choices": [{"message": {"content": "pass"}}]}), ]) with self.assertRaisesRegex(VisualReviewError, "valid review tool call"): self._review(Path(temporary), client) def test_other_http_errors_do_not_retry(self) -> None: with tempfile.TemporaryDirectory() as temporary: client = _Client([_Response(400, text="invalid image")]) with self.assertRaisesRegex(VisualReviewError, r"failed \(400\)"): self._review(Path(temporary), client) self.assertEqual(len(client.requests), 1) def test_candidate_review_requires_complete_coverage_and_achieved_acceptance(self) -> None: with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) image = root / "iso.png" image.write_bytes(b"png") client = _Client([_candidate_response()]) with patch("app.services.visual_review.httpx.AsyncClient", return_value=client): result = asyncio.run(review_candidate_batch( settings(root), manifest={"renderer": "test", "source": "test", "views": [{"id": "isometric", "path": str(image), "camera": {}}]}, requirements="# Frozen requirements", source_requirements="Build a plate with a mounting hole.", checklist=["one connected plate"], batch_goal="Create the connected base plate.", deterministic_report={"health": {"solid_count": 1}}, node_id="candidate_001", )) self.assertEqual(result["verdict"], "accept") self.assertEqual(result["schema_version"], "cad.candidate-review.v1") request = client.requests[0] self.assertEqual(request["tool_choice"], {"type": "function", "function": {"name": "review_candidate_batch"}}) review_input = json.loads(request["messages"][1]["content"][0]["text"]) self.assertEqual(review_input["source_requirements"], "Build a plate with a mounting hole.") self.assertEqual(review_input["frozen_requirements"], "# Frozen requirements") self.assertIn("may not remove, replace, or weaken", review_input["instruction"]) def test_candidate_reviewer_retries_invalid_coverage_on_same_rendered_candidate(self) -> None: with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) image = root / "iso.png" image.write_bytes(b"png") invalid = _candidate_response() invalid._body["choices"][0]["message"]["tool_calls"][0]["function"]["arguments"] = json.dumps({ "verdict": "accept", "confidence": 0.96, "batch_goal_status": "achieved", "coverage": [], "evidence": ["The staged plate achieves the batch goal."], }) client = _Client([invalid, _candidate_response()]) with patch("app.services.visual_review.httpx.AsyncClient", return_value=client): result = asyncio.run(review_candidate_batch( settings(root), manifest={"renderer": "test", "source": "test", "views": [{"id": "isometric", "path": str(image), "camera": {}}]}, requirements="# Frozen requirements", source_requirements="Build a plate.", checklist=["one connected plate"], batch_goal="Create the connected base plate.", deterministic_report={"health": {"solid_count": 1}}, node_id="candidate_001", )) self.assertEqual(result["verdict"], "accept") self.assertEqual(len(client.requests), 2) first_input = client.requests[0]["messages"][1]["content"] second_input = client.requests[1]["messages"][1]["content"] self.assertEqual(second_input, first_input) self.assertIn("must return coverage for every checklist item", client.requests[1]["messages"][2]["content"]) def test_candidate_reviewer_stops_after_one_invalid_output_retry(self) -> None: with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) image = root / "iso.png" image.write_bytes(b"png") invalid_responses = [] for _ in range(2): response = _candidate_response() response._body["choices"][0]["message"]["tool_calls"][0]["function"]["arguments"] = json.dumps({ "verdict": "accept", "confidence": 0.9, "batch_goal_status": "achieved", "coverage": [], "evidence": ["bad coverage"], }) invalid_responses.append(response) client = _Client(invalid_responses) with patch("app.services.visual_review.httpx.AsyncClient", return_value=client): with self.assertRaisesRegex(VisualReviewError, "coverage for every checklist item"): asyncio.run(review_candidate_batch( settings(root), manifest={"renderer": "test", "source": "test", "views": [{"id": "isometric", "path": str(image), "camera": {}}]}, requirements="# Frozen requirements", checklist=["one connected plate"], batch_goal="Create the connected base plate.", deterministic_report={}, node_id="candidate_001", )) self.assertEqual(len(client.requests), 2) def test_candidate_reviewer_cannot_accept_a_partial_batch(self) -> None: with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) image = root / "iso.png" image.write_bytes(b"png") client = _Client([ _candidate_response(batch_goal_status="partial"), _candidate_response(batch_goal_status="partial"), ]) with patch("app.services.visual_review.httpx.AsyncClient", return_value=client): with self.assertRaisesRegex(VisualReviewError, "achieved batch goal"): asyncio.run(review_candidate_batch( settings(root), manifest={"renderer": "test", "source": "test", "views": [{"id": "isometric", "path": str(image), "camera": {}}]}, requirements="# Frozen requirements", checklist=["one connected plate"], batch_goal="Create the connected base plate.", deterministic_report={}, node_id="candidate_001", )) self.assertEqual(len(client.requests), 2) if __name__ == "__main__": unittest.main()