Jev 운영 배포 검증과 남은 확인 항목 기록

scripts/test_probe_jev_dialogue.py

@@ -2,11 +2,13 @@
 
 from __future__ import annotations
 
+import asyncio
 import importlib.util
 import io
 import unittest
 from contextlib import redirect_stderr
 from pathlib import Path
+from types import SimpleNamespace
 
 
 SCRIPT_PATH = Path(__file__).with_name("probe-jev-dialogue.py")
@@ -17,6 +19,35 @@ SPEC.loader.exec_module(probe)
 
 
 class ProbeJevDialogueTest(unittest.TestCase):
+    def _cleanup_runtime(self, response: object, *, live_provider: str = "openai", engine_mode: str = "remote"):
+        requests: list[str] = []
+
+        class Client:
+            async def delete(self, path: str) -> object:
+                requests.append(path)
+                return response
+
+        return {
+            "settings": SimpleNamespace(
+                live_client_provider=live_provider,
+                engine_mode=engine_mode,
+            ),
+            "engine_client": SimpleNamespace(client=Client()),
+        }, requests
+
+    @staticmethod
+    def _phase(provider: str) -> dict[str, object]:
+        return {
+            "status": "done",
+            "turns": [
+                {
+                    "status": "done",
+                    "generation": {"provider": provider},
+                }
+            ],
+            "session_closed": False,
+        }
+
     def test_parser_validates_phase_and_repeat_contract(self) -> None:
         args = probe.build_parser().parse_args(
             ["--output", "report.json", "--phases", "jev,legacy", "--repeats", "3", "--label", "trial-a"]
@@ -85,6 +116,107 @@ class ProbeJevDialogueTest(unittest.TestCase):
             ],
         )
 
+    def test_openai_stateless_cleanup_allows_http_200_closed_false(self) -> None:
+        class Response:
+            status_code = 200
+
+            def raise_for_status(self) -> None:
+                return None
+
+            def json(self) -> dict[str, bool]:
+                return {"closed": False}
+
+        response = Response()
+        runtime, requests = self._cleanup_runtime(response)
+        phase = self._phase("openai")
+
+        asyncio.run(probe._cleanup_synthetic_session(runtime=runtime, phase=phase, session_id="session-1"))
+
+        self.assertEqual(requests, ["/session/session-1"])
+        self.assertEqual(phase["status"], "done")
+        self.assertFalse(phase["session_closed"])
+        self.assertEqual(phase["cleanup_http_status"], 200)
+        self.assertFalse(phase["cleanup_closed"])
+        self.assertEqual(phase["cleanup_status"], "not_required")
+        self.assertNotIn("cleanup_error", phase)
+
+    def test_non_openai_cleanup_closed_false_is_failure(self) -> None:
+        class Response:
+            status_code = 200
+
+            def raise_for_status(self) -> None:
+                return None
+
+            def json(self) -> dict[str, bool]:
+                return {"closed": False}
+
+        response = Response()
+        runtime, requests = self._cleanup_runtime(response)
+        phase = self._phase("claude")
+
+        asyncio.run(probe._cleanup_synthetic_session(runtime=runtime, phase=phase, session_id="session-2"))
+
+        self.assertEqual(requests, ["/session/session-2"])
+        self.assertEqual(phase["status"], "error")
+        self.assertFalse(phase["session_closed"])
+        self.assertEqual(phase["cleanup_http_status"], 200)
+        self.assertFalse(phase["cleanup_closed"])
+        self.assertEqual(phase["cleanup_status"], "failed")
+        self.assertEqual(phase["cleanup_error"], "engine_close_failed")
+
+    def test_openai_cleanup_http_failure_is_failure(self) -> None:
+        class Response:
+            status_code = 503
+
+            def __init__(self) -> None:
+                self.json_called = False
+
+            def raise_for_status(self) -> None:
+                raise RuntimeError("http failure")
+
+            def json(self) -> dict[str, bool]:
+                self.json_called = True
+                return {"closed": False}
+
+        response = Response()
+        runtime, requests = self._cleanup_runtime(response)
+        phase = self._phase("openai")
+
+        asyncio.run(probe._cleanup_synthetic_session(runtime=runtime, phase=phase, session_id="session-3"))
+
+        self.assertEqual(requests, ["/session/session-3"])
+        self.assertEqual(phase["status"], "error")
+        self.assertFalse(phase["session_closed"])
+        self.assertEqual(phase["cleanup_http_status"], 503)
+        self.assertFalse(response.json_called)
+        self.assertNotIn("cleanup_closed", phase)
+        self.assertEqual(phase["cleanup_status"], "failed")
+        self.assertEqual(phase["cleanup_error"], "engine_close_failed")
+
+    def test_openai_cleanup_requires_boolean_closed_json(self) -> None:
+        class Response:
+            status_code = 200
+
+            def raise_for_status(self) -> None:
+                return None
+
+            def json(self) -> dict[str, int]:
+                return {"closed": 0}
+
+        response = Response()
+        runtime, requests = self._cleanup_runtime(response)
+        phase = self._phase("openai")
+
+        asyncio.run(probe._cleanup_synthetic_session(runtime=runtime, phase=phase, session_id="session-4"))
+
+        self.assertEqual(requests, ["/session/session-4"])
+        self.assertEqual(phase["status"], "error")
+        self.assertFalse(phase["session_closed"])
+        self.assertEqual(phase["cleanup_http_status"], 200)
+        self.assertNotIn("cleanup_closed", phase)
+        self.assertEqual(phase["cleanup_status"], "failed")
+        self.assertEqual(phase["cleanup_error"], "engine_close_failed")
+
 
 if __name__ == "__main__":
     unittest.main()