feat(workflow): JoyAI-VL-Interaction tool capability node

.env.example

@@ -58,6 +58,12 @@ DATABASE_URL=sqlite+aiosqlite:///./opencli_admin.db
 #          docker-compose --profile celery up
 TASK_EXECUTOR=local
 
+# JoyAI-VL-Interaction 工具节点（tool.realtime.vl.interaction）
+# 自托管 vLLM-Omni 服务地址（jd-opensource/JoyAI-VL-Interaction, 8B, Apache-2.0）。
+# 留空 = 节点执行降级为 error 事件, 不崩 run。可选 JOYAI_VL_API_KEY 加 Bearer。
+# JOYAI_VL_URL=http://joyai-vl:8000
+# JOYAI_VL_API_KEY=
+
 # 采集编排（NAS 推荐 iii）
 # admin — API 内置 scheduler.py 驱动定时采集（默认）
 # iii   — III engine + schedule-bootstrap 驱动 cron；见 .env.nas.example

backend/schemas/workflow.py

@@ -390,7 +390,7 @@ class WorkflowToolCapabilityPort(BaseModel):
 
 
 class WorkflowToolCapabilityExecutor(BaseModel):
-    mode: Literal["fixture", "okx_market_ticker_snapshot"]
+    mode: Literal["fixture", "okx_market_ticker_snapshot", "joyai_vl_interaction"]
     description: Optional[str] = None
     params: dict[str, Any] = Field(default_factory=dict)
 

backend/workflow/joyai_vl_executor.py

@@ -0,0 +1,148 @@
+"""JoyAI-VL-Interaction executor for OpenCLI Tool Capabilities.
+
+Bridges the workflow runtime to a self-hosted JoyAI-VL-Interaction deployment
+(https://github.com/jd-opensource/JoyAI-VL-Interaction) — JD's 8B real-time
+video-language interaction model, served OpenAI-compatible via vLLM-Omni.
+
+MVP scope: one synchronous interaction probe per tool call — send the node's
+prompt plus media references (video URL / image frames) to the model's
+chat-completions endpoint and emit the reply as an event.v1-style payload.
+The always-on streaming mode (model watches a live feed and speaks up
+unprompted) needs a persistent session and lands later on the same executor.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import time
+import urllib.request
+from datetime import UTC, datetime
+from typing import Any
+
+JOYAI_VL_INTERACTION_EXECUTOR = "joyai_vl_interaction"
+JOYAI_VL_TOOL_CAPABILITY_ID = "tool.realtime.vl.interaction"
+JOYAI_VL_DEFAULT_MODEL = "JoyAI-VL-Interaction-Preview"
+JOYAI_VL_ENDPOINT_ENV = "JOYAI_VL_URL"
+
+
+class JoyAIVLExecutionError(RuntimeError):
+    """Raised when the JoyAI-VL interaction executor cannot produce a reply."""
+
+
+def execute_joyai_vl_interaction(params: dict[str, Any]) -> dict[str, Any]:
+    """Run one vision-language interaction turn against a JoyAI-VL deployment."""
+
+    endpoint = (
+        _read_string(params.get("endpoint"))
+        or _read_string(params.get("endpointUrl"))
+        or _read_string(os.environ.get(JOYAI_VL_ENDPOINT_ENV))
+    )
+    if not endpoint:
+        raise JoyAIVLExecutionError(
+            f"JoyAI-VL endpoint is not configured: set {JOYAI_VL_ENDPOINT_ENV} "
+            "(vLLM-Omni base URL, e.g. http://joyai-vl:8000) or pass params.endpoint"
+        )
+
+    model = _read_string(params.get("model")) or JOYAI_VL_DEFAULT_MODEL
+    prompt = _read_string(params.get("prompt")) or "描述当前画面正在发生什么, 如有需要人工注意的事件请指出."
+    video_url = _read_string(params.get("videoUrl")) or _read_string(params.get("video_url"))
+    image_urls = _read_string_list(params.get("imageUrls")) or _read_string_list(
+        params.get("image_urls")
+    )
+    timeout_seconds = _read_timeout(params.get("timeoutSeconds"))
+
+    content: list[dict[str, Any]] = []
+    if video_url:
+        content.append({"type": "video_url", "video_url": {"url": video_url}})
+    for url in image_urls:
+        content.append({"type": "image_url", "image_url": {"url": url}})
+    content.append({"type": "text", "text": prompt})
+
+    request_body = {
+        "model": model,
+        "messages": [{"role": "user", "content": content}],
+        "stream": False,
+    }
+    max_tokens = _read_int(params.get("maxTokens"))
+    if max_tokens:
+        request_body["max_tokens"] = max_tokens
+
+    url = endpoint.rstrip("/") + "/v1/chat/completions"
+    opened_at = time.time()
+    request = urllib.request.Request(
+        url,
+        data=json.dumps(request_body).encode("utf-8"),
+        headers={
+            "Content-Type": "application/json",
+            "User-Agent": "Mozilla/5.0 OpenCLI-Admin-joyai-vl-executor/0.1",
+            **_auth_header(params),
+        },
+        method="POST",
+    )
+
+    try:
+        with urllib.request.urlopen(request, timeout=timeout_seconds) as response:
+            payload = json.loads(response.read().decode("utf-8"))
+    except JoyAIVLExecutionError:
+        raise
+    except Exception as exc:  # pragma: no cover - exercised by live smoke failures.
+        raise JoyAIVLExecutionError(f"JoyAI-VL request failed: {exc}") from exc
+
+    choices = payload.get("choices") or []
+    if not choices:
+        raise JoyAIVLExecutionError(f"JoyAI-VL returned no choices: {payload}")
+    message = choices[0].get("message") or {}
+    reply = message.get("content")
+    if not isinstance(reply, str) or not reply.strip():
+        raise JoyAIVLExecutionError(f"JoyAI-VL returned an empty reply: {payload}")
+
+    duration_ms = round((time.time() - opened_at) * 1000)
+    return {
+        "schema": "event.vl.interaction.v1",
+        "source": "joyai-vl",
+        "eventType": "vl.interaction",
+        "observedAt": datetime.now(tz=UTC).isoformat(),
+        "model": payload.get("model") or model,
+        "reply": reply,
+        "media": {
+            "videoUrl": video_url,
+            "imageCount": len(image_urls),
+        },
+        "request": {
+            "url": url,
+            "prompt": prompt,
+            "durationMs": duration_ms,
+        },
+        "usage": payload.get("usage") or {},
+    }
+
+
+def _auth_header(params: dict[str, Any]) -> dict[str, str]:
+    api_key = _read_string(params.get("apiKey")) or _read_string(
+        os.environ.get("JOYAI_VL_API_KEY")
+    )
+    return {"Authorization": f"Bearer {api_key}"} if api_key else {}
+
+
+def _read_string(value: object) -> str | None:
+    return value.strip() if isinstance(value, str) and value.strip() else None
+
+
+def _read_string_list(value: object) -> list[str]:
+    if not isinstance(value, list):
+        return []
+    return [item.strip() for item in value if isinstance(item, str) and item.strip()]
+
+
+def _read_int(value: object) -> int | None:
+    try:
+        return int(value) if value is not None else None
+    except (TypeError, ValueError):
+        return None
+
+
+def _read_timeout(value: object) -> float:
+    if isinstance(value, int | float) and value > 0:
+        return float(value)
+    return 30.0

backend/workflow/opencli_hda_tracer.py

@@ -46,6 +46,12 @@
 from backend.workflow.compiler import INTERNAL_ID_SEPARATOR, compile_workflow_project
 from backend.workflow.event_mirror import publish_workflow_run_event_mirror
 from backend.workflow.fleet_inventory import match_workflow_fleet_capability
+from backend.workflow.joyai_vl_executor import (
+    JOYAI_VL_INTERACTION_EXECUTOR,
+    JOYAI_VL_TOOL_CAPABILITY_ID,
+    JoyAIVLExecutionError,
+    execute_joyai_vl_interaction,
+)
 from backend.workflow.realtime_market_executor import (
     OKX_MARKET_TICKER_SNAPSHOT_EXECUTOR,
     RealtimeMarketExecutionError,
@@ -1677,6 +1683,13 @@ def _execute_external_tool_capability(
         output = _execute_okx_market_tool(binding_input)
         return [_external_tool_output(node, output, input_items, run_id, 0, binding_input)]
 
+    if (
+        binding_input.get("executorMode") == JOYAI_VL_INTERACTION_EXECUTOR
+        and binding_input.get("toolCapabilityId") == JOYAI_VL_TOOL_CAPABILITY_ID
+    ):
+        output = _execute_joyai_vl_tool(binding_input)
+        return [_external_tool_output(node, output, input_items, run_id, 0, binding_input)]
+
     fixture_outputs = _read_dict_list(binding_input.get("fixtureOutputs"))
     fixture_output = _read_dict(binding_input.get("fixtureOutput"))
     if not fixture_outputs and fixture_output:
@@ -1707,6 +1720,23 @@ def _execute_okx_market_tool(binding_input: dict[str, Any]) -> dict[str, Any]:
         }
 
 
+def _execute_joyai_vl_tool(binding_input: dict[str, Any]) -> dict[str, Any]:
+    params = {
+        **_read_dict(binding_input.get("executorParams")),
+        **_read_dict(binding_input.get("toolParams")),
+    }
+    try:
+        return execute_joyai_vl_interaction(params)
+    except JoyAIVLExecutionError as exc:
+        return {
+            "schema": "event.vl.interaction.error.v1",
+            "source": "joyai-vl",
+            "eventType": "vl.interaction.error",
+            "status": "error",
+            "message": str(exc),
+        }
+
+
 def _external_tool_output(
     node: CompiledWorkflowNode,
     output: dict[str, Any],

backend/workflow/runtime_registry.py

@@ -51,7 +51,7 @@
 WEBHOOK_NOTIFY_BINDING_ID = "workflow.notifier.webhook.send"
 NOTIFY_SEND_BINDING_ID = "workflow.notify.send"
 EXTERNAL_TOOL_BINDING_ID = "workflow.external-tool.capability"
-SUPPORTED_TOOL_EXECUTOR_MODES = {"fixture", "okx_market_ticker_snapshot"}
+SUPPORTED_TOOL_EXECUTOR_MODES = {"fixture", "okx_market_ticker_snapshot", "joyai_vl_interaction"}
 
 
 class WorkflowRuntimeBinding(BaseModel):

backend/workflow/tool_capabilities.py

@@ -8,6 +8,10 @@
     WorkflowToolCapabilityExecutor,
     WorkflowToolCapabilityPort,
 )
+from backend.workflow.joyai_vl_executor import (
+    JOYAI_VL_INTERACTION_EXECUTOR,
+    JOYAI_VL_TOOL_CAPABILITY_ID,
+)
 from backend.workflow.realtime_market_executor import OKX_MARKET_TICKER_SNAPSHOT_EXECUTOR
 
 
@@ -141,6 +145,30 @@ def _tool_capabilities() -> list[WorkflowToolCapability]:
             schema="tool-capability.realtime-signal-emit.v1",
             resources=["signal_policy", "run_trace"],
         ),
+        _realtime_tool(
+            id=JOYAI_VL_TOOL_CAPABILITY_ID,
+            label="JoyAI VL Interaction",
+            description=(
+                "Vision-language interaction over video/image media via a "
+                "self-hosted JoyAI-VL-Interaction deployment (JD open source, "
+                "8B, vLLM-Omni serving). Sends the node prompt plus media "
+                "references, emits the model's interaction reply as event[]."
+            ),
+            input_type="event[]",
+            output_type="event[]",
+            tags=["tool", "realtime", "vision", "vl", "interaction", "joyai"],
+            schema="tool-capability.vl-interaction.v1",
+            resources=["media_reference", "run_trace"],
+            executor=WorkflowToolCapabilityExecutor(
+                mode=JOYAI_VL_INTERACTION_EXECUTOR,
+                description=(
+                    "POSTs one chat-completions interaction turn to the "
+                    "JOYAI_VL_URL vLLM-Omni endpoint; unconfigured endpoint "
+                    "degrades to an error event, never a run crash."
+                ),
+                params={"model": "JoyAI-VL-Interaction-Preview"},
+            ),
+        ),
     ]
 
 

docs/BUG-TRIAGE-20260718.md

@@ -51,3 +51,11 @@
 cd D:\projects\opencli-admin
 $env:API_AUTH_TOKEN=''; $env:AGENT_API_TOKEN=''; uv run --extra dev pytest tests/unit tests/integration -q --no-cov
 ```
+
+## P3 补录 (2026-07-18 晚)
+
+### 6. 全套跑测试顺序污染 flake (main 存量)
+- `pytest tests/integration tests/unit tests/skills` 全套跑稳定挂 6 个, 单跑全过; clean main 与 feature 分支同样 6 个 → 存量顺序依赖, 非新代码
+- 名单: test_workflow_opencli_hda_trace_api (projection/stream ×2), test_workflow_turbopush_publish_api (capabilities ×1), test_nodes_install_script (netbird/ssh ×2), +1
+- 疑似共享进程级状态 (capability registry / settings snapshot / os.environ 写入未回滚)
+- 修向: 定位先污染后断言的 test 对 (pytest -p no:randomly 二分), 或对相关 fixture 加隔离

tests/unit/test_joyai_vl_executor.py

@@ -0,0 +1,101 @@
+"""JoyAI-VL interaction tool executor: config guard, request shape, reply parsing."""
+
+import io
+import json
+
+import pytest
+
+from backend.workflow.joyai_vl_executor import (
+    JOYAI_VL_ENDPOINT_ENV,
+    JOYAI_VL_TOOL_CAPABILITY_ID,
+    JoyAIVLExecutionError,
+    execute_joyai_vl_interaction,
+)
+from backend.workflow.tool_capabilities import (
+    list_workflow_tool_capabilities,
+    resolve_workflow_tool_capability,
+)
+
+
+def test_unconfigured_endpoint_raises_actionable_error(monkeypatch):
+    monkeypatch.delenv(JOYAI_VL_ENDPOINT_ENV, raising=False)
+    with pytest.raises(JoyAIVLExecutionError, match=JOYAI_VL_ENDPOINT_ENV):
+        execute_joyai_vl_interaction({})
+
+
+def test_success_turn_builds_openai_media_request_and_parses_reply(monkeypatch):
+    captured: dict = {}
+
+    class _FakeResponse(io.BytesIO):
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *args):
+            return False
+
+    def fake_urlopen(request, timeout):
+        captured["url"] = request.full_url
+        captured["timeout"] = timeout
+        captured["body"] = json.loads(request.data.decode("utf-8"))
+        return _FakeResponse(
+            json.dumps(
+                {
+                    "model": "JoyAI-VL-Interaction-Preview",
+                    "choices": [{"message": {"role": "assistant", "content": "锅要溢出来了"}}],
+                    "usage": {"total_tokens": 42},
+                }
+            ).encode("utf-8")
+        )
+
+    monkeypatch.setattr("urllib.request.urlopen", fake_urlopen)
+
+    output = execute_joyai_vl_interaction(
+        {
+            "endpoint": "http://joyai-vl:8000/",
+            "prompt": "看着炉子",
+            "videoUrl": "http://media/stove.mp4",
+            "imageUrls": ["http://media/frame1.jpg"],
+            "timeoutSeconds": 5,
+        }
+    )
+
+    assert captured["url"] == "http://joyai-vl:8000/v1/chat/completions"
+    assert captured["timeout"] == 5.0
+    content = captured["body"]["messages"][0]["content"]
+    assert {"type": "video_url", "video_url": {"url": "http://media/stove.mp4"}} in content
+    assert {"type": "image_url", "image_url": {"url": "http://media/frame1.jpg"}} in content
+    assert content[-1] == {"type": "text", "text": "看着炉子"}
+
+    assert output["schema"] == "event.vl.interaction.v1"
+    assert output["source"] == "joyai-vl"
+    assert output["reply"] == "锅要溢出来了"
+    assert output["media"] == {"videoUrl": "http://media/stove.mp4", "imageCount": 1}
+    assert output["usage"] == {"total_tokens": 42}
+
+
+def test_empty_choices_raises(monkeypatch):
+    class _FakeResponse(io.BytesIO):
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *args):
+            return False
+
+    monkeypatch.setattr(
+        "urllib.request.urlopen",
+        lambda request, timeout: _FakeResponse(json.dumps({"choices": []}).encode("utf-8")),
+    )
+    with pytest.raises(JoyAIVLExecutionError, match="no choices"):
+        execute_joyai_vl_interaction({"endpoint": "http://joyai-vl:8000"})
+
+
+def test_tool_capability_is_registered_and_resolvable():
+    tool = resolve_workflow_tool_capability(JOYAI_VL_TOOL_CAPABILITY_ID)
+    assert tool is not None
+    assert tool.executor.mode == "joyai_vl_interaction"
+    assert tool.status == "runnable"
+    assert [port.type for port in tool.inputPorts] == ["event[]"]
+    assert [port.type for port in tool.outputPorts] == ["event[]"]
+    assert JOYAI_VL_TOOL_CAPABILITY_ID in {
+        t.id for t in list_workflow_tool_capabilities().tools
+    }