ff

MPV/pipeline_helper.py

@@ -28,6 +28,7 @@ import os
 import sys
 import tempfile
 import time
+import threading
 import logging
 import re
 import hashlib
@@ -134,8 +135,16 @@ def _run_pipeline(pipeline_text: str, *, seeds: Any = None) -> Dict[str, Any]:
             "rows": rows_payload
         }
 
+    start_time = time.time()
     runner = PipelineRunner()
     result = runner.run_pipeline(pipeline_text, seeds=seeds)
+    duration = time.time() - start_time
+    try:
+        _append_helper_log(
+            f"[pipeline] run_pipeline completed in {duration:.2f}s pipeline={pipeline_text[:64]}"
+        )
+    except Exception:
+        pass
 
     table_payload = None
     try:
@@ -152,6 +161,31 @@ def _run_pipeline(pipeline_text: str, *, seeds: Any = None) -> Dict[str, Any]:
     }
 
 
+def _run_pipeline_background(pipeline_text: str, *, seeds: Any, req_id: str) -> None:
+    def _target() -> None:
+        try:
+            result = _run_pipeline(pipeline_text, seeds=seeds)
+            status = "success" if result.get("success") else "failed"
+            _append_helper_log(
+                f"[pipeline async {req_id}] {status} error={result.get('error')}"
+            )
+        except Exception as exc:  # pragma: no cover - best-effort logging
+            _append_helper_log(
+                f"[pipeline async {req_id}] exception: {type(exc).__name__}: {exc}"
+            )
+
+    thread = threading.Thread(
+        target=_target,
+        name=f"pipeline-async-{req_id}",
+        daemon=True,
+    )
+    thread.start()
+
+
+def _is_load_url_pipeline(pipeline_text: str) -> bool:
+    return str(pipeline_text or "").lstrip().lower().startswith(".mpv -url")
+
+
 def _run_op(op: str, data: Any) -> Dict[str, Any]:
     """Run a helper-only operation.
 
@@ -1030,13 +1064,29 @@ def main(argv: Optional[list[str]] = None) -> int:
         except Exception:
             pass
 
+        async_dispatch = False
         try:
             if op:
                 run = _run_op(op, data)
             else:
                 if not pipeline_text:
                     continue
-                run = _run_pipeline(pipeline_text, seeds=seeds)
+                if _is_load_url_pipeline(pipeline_text):
+                    async_dispatch = True
+                    run = {
+                        "success": True,
+                        "stdout": "",
+                        "stderr": "",
+                        "error": "",
+                        "table": None,
+                    }
+                    _run_pipeline_background(
+                        pipeline_text,
+                        seeds=seeds,
+                        req_id=req_id,
+                    )
+                else:
+                    run = _run_pipeline(pipeline_text, seeds=seeds)
 
             resp = {
                 "id": req_id,
@@ -1050,6 +1100,8 @@ def main(argv: Optional[list[str]] = None) -> int:
             }
             if "choices" in run:
                 resp["choices"] = run.get("choices")
+            if async_dispatch:
+                resp["info"] = "queued asynchronously"
         except Exception as exc:
             resp = {
                 "id": req_id,