Stabilize skill quality reporter async wrapper

openspace/cloud/skill_quality_reporter.py

@@ -3,6 +3,8 @@
 from __future__ import annotations
 
 import asyncio
+from concurrent.futures import Future as ConcurrentFuture
+from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
 from typing import Any
 
@@ -28,6 +30,7 @@ logger = Logger.get_logger(__name__)
 QUALITY_EVENT_KIND = "skill_judgment"
 QUALITY_SCHEMA_VERSION = "skill_quality_v1"
 QUALITY_DENOMINATOR = "analyzer_judged_skill_use"
+EXECUTOR_POLL_INTERVAL_SECONDS = 0.01
 
 
 class CloudSkillQualityReporter:
@@ -57,12 +60,39 @@ class CloudSkillQualityReporter:
         *,
         session_id: str | None = None,
     ) -> dict[str, Any]:
+        executor: ThreadPoolExecutor | None = None
+        result_future: ConcurrentFuture[dict[str, Any]] | None = None
+        loop_future: asyncio.Future[Any] | None = None
         try:
-            return await asyncio.to_thread(
-                self._maybe_report_analysis_sync,
-                analysis,
-                session_id=session_id,
+            if not load_cloud_skill_quality_reporting_enabled():
+                return _skill_quality_reporting_disabled_result()
+
+            executor = ThreadPoolExecutor(
+                max_workers=1,
+                thread_name_prefix="openspace-skill-quality-reporter",
             )
+            result_future = ConcurrentFuture()
+
+            def run_sync_report() -> None:
+                if not result_future.set_running_or_notify_cancel():
+                    return
+                try:
+                    result_future.set_result(
+                        self._maybe_report_analysis_sync(
+                            analysis,
+                            session_id=session_id,
+                        )
+                    )
+                except BaseException as exc:
+                    result_future.set_exception(exc)
+
+            loop = asyncio.get_running_loop()
+            loop_future = loop.run_in_executor(executor, run_sync_report)
+            while not result_future.done():
+                await asyncio.sleep(EXECUTOR_POLL_INTERVAL_SECONDS)
+            if not loop_future.done():
+                loop_future.cancel()
+            return result_future.result()
         except Exception as exc:
             logger.warning(
                 "Skill quality telemetry reporter failed for task %s: %s",
@@ -74,6 +104,12 @@ class CloudSkillQualityReporter:
                 "reason": "reporter_error",
                 "error": type(exc).__name__,
             }
+        finally:
+            if loop_future is not None and not loop_future.done():
+                loop_future.cancel()
+            if executor is not None:
+                work_finished = result_future is not None and result_future.done()
+                executor.shutdown(wait=work_finished, cancel_futures=True)
 
     def _maybe_report_analysis_sync(
         self,
@@ -81,9 +117,6 @@ class CloudSkillQualityReporter:
         *,
         session_id: str | None = None,
     ) -> dict[str, Any]:
-        if not load_cloud_skill_quality_reporting_enabled():
-            return {"status": "skipped", "reason": "skill_quality_reporting_disabled"}
-
         cfg = load_cloud_config()
         if not cfg.enabled or cfg.telemetry_mode != "outbox" or not cfg.api_key:
             return {"status": "skipped", "reason": "cloud_telemetry_disabled"}
@@ -229,6 +262,10 @@ def _analysis_timestamp_iso(analysis: ExecutionAnalysis) -> str:
     return str(timestamp)
 
 
+def _skill_quality_reporting_disabled_result() -> dict[str, Any]:
+    return {"status": "skipped", "reason": "skill_quality_reporting_disabled"}
+
+
 def _summarize_outcomes(outcomes: list[dict[str, Any]]) -> dict[str, Any]:
     if any(item.get("status") == "reported" for item in outcomes):
         status = "reported"

tests/test_skill_quality_reporter.py

@@ -234,8 +234,24 @@ def test_disabled_quality_gate_skips_before_invalid_cloud_config(monkeypatch, tm
     def fail_client_construction(*args, **kwargs):
         raise AssertionError("OpenSpaceClient should not be constructed")
 
+    def fail_config_load(*args, **kwargs):
+        raise AssertionError("load_cloud_config should not be called")
+
+    def fail_outbox_construction(*args, **kwargs):
+        raise AssertionError("CloudTelemetryOutbox should not be constructed")
+
+    def fail_executor_construction(*args, **kwargs):
+        raise AssertionError("ThreadPoolExecutor should not be constructed")
+
+    def fail_sync_body(*args, **kwargs):
+        raise AssertionError("_maybe_report_analysis_sync should not be called")
+
+    monkeypatch.setattr(reporter_module, "load_cloud_config", fail_config_load)
     monkeypatch.setattr(reporter_module, "OpenSpaceClient", fail_client_construction)
+    monkeypatch.setattr(reporter_module, "CloudTelemetryOutbox", fail_outbox_construction)
+    monkeypatch.setattr(reporter_module, "ThreadPoolExecutor", fail_executor_construction)
     reporter = CloudSkillQualityReporter(workspace_root=tmp_path)
+    monkeypatch.setattr(reporter, "_maybe_report_analysis_sync", fail_sync_body)
 
     result = _run(reporter.maybe_report_analysis(_analysis()))