Pipeline maintainence (#451)

Author: ultmaster

.github/workflows/tests.yml

@@ -57,10 +57,13 @@ jobs:
         uses: pre-commit/action@v3.0.1
       - name: Check Python headers
         run: uv run --locked --no-sync scripts/check_headers.py
+        if: matrix.setup == 'fast'
       - name: Run Black
         run: uv run --locked --no-sync black --check .
+        if: matrix.setup != 'next'
       - name: Run isort
         run: uv run --locked --no-sync isort --check-only .
+        if: matrix.setup != 'next'
       - name: Run pyright (fast)
         run: uv run --locked --no-sync pyright -p pyrightconfig.fast.json
         if: matrix.setup == 'fast'

agentlightning/emitter/annotation.py

@@ -246,7 +246,11 @@ def _record_auto_outputs(recording_ctx: SpanRecordingContext, result: Any) -> No
             flattened = flatten_attributes({LightningSpanAttributes.OPERATION_OUTPUT.value: result})
             recording_ctx.record_attributes(sanitize_attributes(flattened))
 
-        if asyncio.iscoroutinefunction(fn) or inspect.iscoroutinefunction(fn):
+        if inspect.iscoroutinefunction(fn) or (
+            # For backwards compatibility.
+            hasattr(asyncio, "iscoroutinefunction")
+            and asyncio.iscoroutinefunction(fn)  # type: ignore
+        ):
 
             @functools.wraps(fn)
             async def async_wrapper(*args: Any, **kwargs: Any) -> Any:

agentlightning/emitter/reward.py

@@ -92,7 +92,11 @@ def wrap_result(result: Optional[float]) -> _RewardSpanData:
         return {"type": "reward", "value": float(result)}
 
     # Check if the function is async
-    is_async = asyncio.iscoroutinefunction(fn) or inspect.iscoroutinefunction(fn)
+    is_async = inspect.iscoroutinefunction(fn) or (
+        # For backwards compatibility.
+        hasattr(asyncio, "iscoroutinefunction")
+        and asyncio.iscoroutinefunction(fn)  # type: ignore
+    )
 
     if is_async:
 

agentlightning/instrumentation/weave.py

@@ -424,6 +424,29 @@ def score_list(self, req: tsi.ScoreListReq) -> Iterator[tsi.ScoreReadRes]:
     def score_delete(self, req: tsi.ScoreDeleteReq) -> tsi.ScoreDeleteRes:
         return tsi.ScoreDeleteRes(num_deleted=0)
 
+    # Experimental unstable APIs
+    # We don't support these APIs yet.
+    def annotation_queue_create(self, *args: Any, **kwargs: Any) -> Any:
+        raise NotImplementedError()
+
+    def annotation_queues_query_stream(self, *args: Any, **kwargs: Any) -> Any:
+        raise NotImplementedError()
+
+    def annotation_queue_read(self, *args: Any, **kwargs: Any) -> Any:
+        raise NotImplementedError()
+
+    def annotation_queue_add_calls(self, *args: Any, **kwargs: Any) -> Any:
+        raise NotImplementedError()
+
+    def annotation_queues_stats(self, *args: Any, **kwargs: Any) -> Any:
+        raise NotImplementedError()
+
+    def annotation_queue_items_query(self, *args: Any, **kwargs: Any) -> Any:
+        raise NotImplementedError()
+
+    def annotator_queue_items_progress_update(self, *args: Any, **kwargs: Any) -> Any:
+        raise NotImplementedError()
+
 
 # Module-level storage for originals
 _original_init_weave_get_server: Callable[..., Any] | None = None

agentlightning/store/collection/mongo.py

@@ -663,6 +663,9 @@ async def upsert(self, items: Sequence[T_model], update_fields: Sequence[str] |
                     return_document=ReturnDocument.AFTER,
                 )
 
+            if result_doc is None:  # pyright: ignore[reportUnnecessaryComparison]
+                raise RuntimeError(f"Upsert resulted in no document for filter: {pk_filter}")
+
             # Because upsert=True, result_doc is guaranteed to be not None
             new_item = self._model_validate_item(result_doc)
             upserted_items.append(new_item)

scripts/cleanup_aoai.py

@@ -0,0 +1,87 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+import os
+
+import requests
+from openai import OpenAI
+
+# Most common Azure OpenAI setup:
+#   AZURE_OPENAI_ENDPOINT="https://<resource>.openai.azure.com"
+#   AZURE_OPENAI_API_KEY="..."
+# Optional (only if your endpoint requires it):
+#   AZURE_OPENAI_API_VERSION="2025-xx-xx"
+#
+# This script treats "delete finetune job" as "cancel finetune job"
+# because fine-tune jobs are typically cancellable, not deletable.
+
+
+def _client() -> OpenAI:
+    # This script assumes AZURE_OPENAI_ENDPOINT and AZURE_OPENAI_API_KEY are set in the environment.
+    endpoint = os.environ["AZURE_OPENAI_ENDPOINT"]
+    api_key = os.environ["AZURE_OPENAI_API_KEY"]
+    return OpenAI(api_key=api_key, base_url=endpoint)
+
+
+def list_data_files():
+    c = _client()
+    return c.files.list(limit=100)
+
+
+def list_finetune_jobs():
+    c = _client()
+    return c.fine_tuning.jobs.list(limit=100)
+
+
+def delete_data_file(file_id: str):
+    c = _client()
+    return c.files.delete(file_id)
+
+
+def cancel_finetune_job(job_id: str):
+    c = _client()
+    return c.fine_tuning.jobs.cancel(job_id)
+
+
+def delete_finetune_job(job_id: str):
+    # This script assumes AZURE_OPENAI_ENDPOINT and AZURE_OPENAI_API_KEY are set in the environment.
+    endpoint = os.environ["AZURE_OPENAI_ENDPOINT"].rstrip("/")
+    api_key = os.environ["AZURE_OPENAI_API_KEY"]
+    root = endpoint.split("/openai")[0]
+
+    url = f"{root}/openai/fine_tuning/jobs/{job_id}"
+    params = {"api-version": os.environ["AZURE_OPENAI_API_VERSION"]}
+
+    resp = requests.delete(url, headers={"api-key": api_key}, params=params, timeout=60)
+    resp.raise_for_status()
+    return resp.content
+
+
+if __name__ == "__main__":
+    # Quick demo: print IDs you could delete
+    jobs = list_finetune_jobs().data
+    files = list_data_files().data
+
+    print("JOBS:")
+    for j in jobs:
+        print(f"  {j.id}  {getattr(j, 'status', '')}  {getattr(j, 'model', '')}")
+
+    print("\nFILES:")
+    for f in files:
+        print(f"  {f.id}  {getattr(f, 'filename', '')}  {getattr(f, 'status', '')}")
+
+    # Delete them all WITHOUT CONFIRMATION!
+    for j in jobs:
+        print(f"Deleting job {j.id}")
+        try:
+            if j.status == "running":
+                cancel_finetune_job(j.id)
+            delete_finetune_job(j.id)
+        except Exception as exc:
+            print(f"  Error deleting job {j.id}: {exc}")
+
+    for f in files:
+        print(f"Deleting file {f.id}")
+        try:
+            delete_data_file(f.id)
+        except Exception as exc:
+            print(f"  Error deleting file {f.id}: {exc}")