feat(xdl): add optional XDL workflow bridge

docs/user_guide/xdl_bridge.md

@@ -0,0 +1,45 @@
+# XDL Bridge
+
+`unilabos.xdl_bridge` 是可选模块，用于把 AI 或用户生成的 XDL 转为标准
+Uni-Lab workflow。它不会改变已有 workflow、设备调度或驱动行为。
+
+## 使用流程
+
+1. 用户与 Agent 确认要使用的实验室设备。
+2. Agent 生成 XDL，并选择该工站对应的 bridge profile。
+3. Agent 调用 `build_xdl_workflow()` 或 `upload_xdl_workflow()`。
+4. workflow 上传到玻尔跃迁后，用户在 Workflow 页面检查并手动启动。
+
+启动 Uni-Lab edge 仍使用原有命令。例如 comprehensive 预设工站：
+
+```bash
+unilab -g unilabos/test/experiments/comprehensive_protocol/comprehensive_station.json \
+  --upload_registry \
+  --addr https://leap-lab.bohrium.com/api/v1 \
+  --disable_browser
+```
+
+AK/SK 应通过命令行、环境变量或会话注入，不能写入 XDL、profile 或日志。
+
+## Python API
+
+```python
+from unilabos.xdl_bridge import build_xdl_workflow, upload_xdl_workflow
+
+profile = "unilabos/test/experiments/comprehensive_protocol/xdl_bridge.yaml"
+workflow = build_xdl_workflow("experiment.xdl", profile)
+result = upload_xdl_workflow("experiment.xdl", profile, tags=["xdl"])
+```
+
+## Profile
+
+Profile 只绑定目标工站：
+
+- 设备图与 registry；
+- 工作站 ID；
+- XDL 硬件角色到工站资源 ID 的映射；
+- `virtual` 或 `real` 运行模式。
+
+XDL 操作到 `TransferProtocol`、`HeatChillProtocol` 等 Uni-Lab Protocol 的映射是
+模块共享合同，不随工站复制。若工站缺少某个 Protocol、handle 或资源绑定，bridge 在
+上传前报错。

tests/xdl_bridge/test_bridge.py

@@ -0,0 +1,52 @@
+from pathlib import Path
+
+from unilabos.xdl_bridge import build_xdl_workflow, load_station_profile
+
+
+ROOT = Path(__file__).parents[2]
+COMPREHENSIVE_PROFILE = (
+    ROOT
+    / "unilabos"
+    / "test"
+    / "experiments"
+    / "comprehensive_protocol"
+    / "xdl_bridge.yaml"
+)
+
+
+def test_comprehensive_profile_uses_shared_protocol_contract():
+    profile = load_station_profile(COMPREHENSIVE_PROFILE)
+
+    assert profile.workstation_id == "OrganicSynthesisStation"
+    assert profile.operation("Transfer")["template"] == "PumpTransferProtocol"
+    assert profile.bind_component("reactor", "reactor") == "main_reactor"
+    assert profile.operation("FilterThrough")["overrides"]["filter_through"] == "filter_1"
+    assert profile.operation("RunColumn")["overrides"]["column"] == "column_1"
+
+
+def test_xdl_builds_standard_unilab_workflow_for_selected_station(tmp_path):
+    xdl = tmp_path / "transfer.xdl"
+    xdl.write_text(
+        """<?xdl version="2.0.0" ?>
+<XDL><Synthesis>
+  <Hardware>
+    <Component id="reactor" type="reactor"/>
+    <Component id="separator" type="separator"/>
+  </Hardware>
+  <Reagents />
+  <Procedure>
+    <Transfer from_vessel="reactor" to_vessel="separator" volume="5 mL"/>
+  </Procedure>
+</Synthesis></XDL>
+""",
+        encoding="utf-8",
+    )
+
+    workflow = build_xdl_workflow(xdl, COMPREHENSIVE_PROFILE, name="transfer")
+
+    node = workflow["nodes"][0]
+    assert node["resource_name"] == "workstation"
+    assert node["device_name"] == "OrganicSynthesisStation"
+    assert node["template_name"] == "PumpTransferProtocol"
+    assert node["param"]["from_vessel"] == "main_reactor"
+    assert node["param"]["to_vessel"] == "separator_1"

unilabos/test/experiments/comprehensive_protocol/xdl_bridge.yaml

@@ -0,0 +1,39 @@
+station:
+  graph: comprehensive_station.json
+  registry: ../../../registry/devices/work_station.yaml
+  workstation_id: OrganicSynthesisStation
+  resource_name: workstation
+  mode: virtual
+
+hardware:
+  ids:
+    reactor: main_reactor
+    separator: separator_1
+    rotavap: rotavap_1
+    filter: filter_1
+    flask_organic: collection_bottle_1
+    recryst_flask: collection_bottle_2
+    distill_flask: collection_bottle_2
+    flask_product: collection_bottle_3
+    dropping_funnel_a: reagent_bottle_4
+    dropping_funnel_b: reagent_bottle_5
+    waste: waste_bottle_1
+    column: column_1
+    silica_gel: column_1
+  types:
+    reactor: main_reactor
+    separator: separator_1
+    rotavap: rotavap_1
+    filter: filter_1
+    flask: collection_bottle_1
+    column: column_1
+
+operation_overrides:
+  Transfer:
+    template: PumpTransferProtocol
+  FilterThrough:
+    overrides:
+      filter_through: filter_1
+  RunColumn:
+    overrides:
+      column: column_1

unilabos/xdl_bridge/__init__.py

@@ -0,0 +1,63 @@
+"""Optional XDL-to-Uni-Lab workflow bridge.
+
+The bridge translates portable XDL into the existing Uni-Lab workflow contract.
+It does not start devices or alter native workflow execution.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+from .builder import build_workflow, validate_workflow
+from .parser import parse_xdl
+from .profile import StationProfile, load_station_profile
+
+
+def build_xdl_workflow(
+    xdl_path: str | Path, profile_path: str | Path, *, name: str | None = None
+) -> dict[str, Any]:
+    procedure = parse_xdl(xdl_path)
+    profile = load_station_profile(profile_path)
+    payload = build_workflow(procedure, profile, name=name or Path(xdl_path).stem)
+    validate_workflow(payload, profile)
+    return payload
+
+
+def upload_xdl_workflow(
+    xdl_path: str | Path,
+    profile_path: str | Path,
+    *,
+    name: str | None = None,
+    tags: list[str] | None = None,
+    description: str = "",
+    client: Any = None,
+) -> dict[str, Any]:
+    payload = build_xdl_workflow(xdl_path, profile_path, name=name)
+    if client is None:
+        from unilabos.app.web import http_client as client
+    workflow_name = name or Path(xdl_path).stem
+    response = client.workflow_import(
+        name=workflow_name,
+        workflow_uuid=payload["workflow_uuid"],
+        workflow_name=workflow_name,
+        nodes=payload["nodes"],
+        edges=payload["edges"],
+        tags=tags or [],
+        description=description,
+        published=False,
+    )
+    if response.get("code") != 0:
+        raise RuntimeError(f"Workflow upload failed: {response}")
+    return response
+
+
+__all__ = [
+    "StationProfile",
+    "build_xdl_workflow",
+    "build_workflow",
+    "load_station_profile",
+    "parse_xdl",
+    "upload_xdl_workflow",
+    "validate_workflow",
+]

unilabos/xdl_bridge/builder.py

@@ -0,0 +1,139 @@
+from __future__ import annotations
+
+from copy import deepcopy
+import json
+import uuid
+from typing import Any
+
+from .models import CanonicalProcedure
+from .profile import StationProfile
+
+
+class BridgeValidationError(ValueError):
+    pass
+
+
+_VESSEL_KEYS = {
+    "vessel",
+    "from_vessel",
+    "to_vessel",
+    "separation_vessel",
+    "filtrate_vessel",
+    "waste_phase_to_vessel",
+    "product_vessel",
+    "waste_vessel",
+}
+
+
+def _normalize_scalar(key: str, value: Any) -> Any:
+    if not isinstance(value, str):
+        return value
+    if value == "true":
+        return True
+    if value == "false":
+        return False
+    if key == "repeats":
+        try:
+            return int(value)
+        except ValueError:
+            return value
+    return value
+
+
+def _edge(source: str, target: str, source_handle: str, target_handle: str) -> dict[str, str]:
+    return {
+        "source": source,
+        "target": target,
+        "source_node_uuid": source,
+        "target_node_uuid": target,
+        "source_handle_key": source_handle,
+        "source_handle_io": "source",
+        "target_handle_key": target_handle,
+        "target_handle_io": "target",
+    }
+
+
+def build_workflow(
+    procedure: CanonicalProcedure, profile: StationProfile, *, name: str
+) -> dict[str, Any]:
+    bindings = {
+        component["id"]: profile.bind_component(
+            component["id"], component.get("type", "")
+        )
+        for component in procedure.components
+    }
+    nodes: list[dict[str, Any]] = []
+    edges: list[dict[str, str]] = []
+    latest_output: dict[str, tuple[str, str]] = {}
+    previous_node: str | None = None
+    for step in procedure.steps:
+        operation = profile.operation(step.operation)
+        node_id = str(uuid.uuid4())
+        parameters = deepcopy(step.parameters)
+        for old, new in operation.get("parameter_aliases", {}).items():
+            if old in parameters and new not in parameters:
+                parameters[new] = parameters.pop(old)
+        for key, value in operation.get("defaults", {}).items():
+            parameters.setdefault(key, value)
+        parameters.update(operation.get("overrides", {}))
+        for key, value in tuple(parameters.items()):
+            if key in _VESSEL_KEYS and isinstance(value, str):
+                try:
+                    parameters[key] = value if value in profile.graph_nodes else bindings[value]
+                except KeyError as exc:
+                    raise BridgeValidationError(
+                        f"{step.source_path}: unbound vessel {value!r}"
+                    ) from exc
+            else:
+                parameters[key] = _normalize_scalar(key, value)
+        nodes.append(
+            {
+                "uuid": node_id,
+                "name": f"Step {step.sequence}",
+                "type": "ILab",
+                "lab_node_type": "ILab",
+                "template_name": operation["template"],
+                "resource_name": profile.resource_name,
+                "device_name": profile.workstation_id,
+                "description": f"{step.operation} operation",
+                "footer": f"{operation['template']}-{profile.resource_name}",
+                "param": parameters,
+            }
+        )
+        for parameter, handle in operation.get("inputs", {}).items():
+            resource_id = parameters.get(parameter)
+            if isinstance(resource_id, str) and resource_id in latest_output:
+                source, source_handle = latest_output[resource_id]
+                edges.append(_edge(source, node_id, source_handle, handle))
+        if previous_node is not None:
+            edges.append(_edge(previous_node, node_id, "ready", "ready"))
+        for parameter, handle in operation.get("outputs", {}).items():
+            resource_id = parameters.get(parameter)
+            if isinstance(resource_id, str):
+                latest_output[resource_id] = (node_id, handle)
+        previous_node = node_id
+    return {
+        "workflow_uuid": str(uuid.uuid4()),
+        "workflow_name": name,
+        "directed": True,
+        "multigraph": False,
+        "graph": {},
+        "nodes": nodes,
+        "edges": edges,
+        "links": edges,
+    }
+
+
+def validate_workflow(payload: dict[str, Any], profile: StationProfile) -> None:
+    serialized = json.dumps(payload)
+    for value in ("PRCXI", "liquid_handler.prcxi", "[WARN:", "device."):
+        if value in serialized:
+            raise BridgeValidationError(f"Forbidden workflow value: {value}")
+    for node in payload.get("nodes", []):
+        if node.get("resource_name") != profile.resource_name:
+            raise BridgeValidationError("Unexpected workflow resource")
+        if node.get("device_name") != profile.workstation_id:
+            raise BridgeValidationError("Unexpected workflow device")
+        for key in _VESSEL_KEYS:
+            if key in node.get("param", {}) and node["param"][key] not in profile.graph_nodes:
+                raise BridgeValidationError(f"Unbound resource {node['name']}.{key}")