feat(extension): TL 接线——executor 适配/validate 帧/form_defaults/context 注入

- extensions.py: ExtensionExecutor 占位改为适配 Dev-B extension_exec
  (ExtensionRunSpec→ExtensionRunRequest 翻译;inject_real_executor 注入)
- validate 两落点补 extension_validate 帧(ready_to_publish+form_defaults,
  Dev-C 发布卡依赖);新增 _confirm_form_defaults
- context.py: init_context 尾部注入真实执行器,失败静默降级 503

全量测试:bioflow_service 324/324、内核 15/15、ui 122/122+373/373、build 通过。

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

bioflow_service/app/context.py

@@ -161,6 +161,19 @@ async def init_context(db_path: str | None = None) -> Context:
         database = await init_db()
     flow_store = init_flow_store(get_settings().flow_docs_dir_abs)
     context = build_context(db=database, flow_store=flow_store)
+    # TL 接线(2026-09-12):扩展执行器接 Dev-B extension_exec。
+    # 失败(如 docker 不可用/配置缺失)静默降级——trial/run 返回 503 而非 500。
+    try:
+        from .services.bioflow import extensions as ext_mod
+        from .services.bioflow.extension_exec import ExtensionExecConfig, ExtensionExecutor as RealExecutor
+        cfg = ExtensionExecConfig()
+        real = RealExecutor(cfg)
+        store = ext_mod.active_store()
+        if store is not None:
+            ext_mod.inject_real_executor(real, store)
+            logger.info("extension executor wired (image suffix=%s)", cfg.docker_tag_suffix)
+    except Exception as exc:  # noqa: BLE001
+        logger.warning("extension executor wiring skipped: %s", exc)
     return context
 
 

bioflow_service/app/services/bioflow/extensions.py

@@ -121,7 +121,7 @@ def _merge(target: dict[str, Any], patch: dict[str, Any]) -> None:
         target[key] = value
 
 
-async def _call_executor(executor: Any, spec_in: ExtensionRunSpec) -> Any:
+async def _call_executor(executor: Any, spec_in: Any) -> Any:
     """执行器统一调用：兼容 async 方法/协程函数/同步函数三类实现。
 
     Dev-B 的真实执行器为 async（docker subprocess）；测试 fake 可 sync。
@@ -183,17 +183,40 @@ class ExtensionRunResult:
 
 
 class ExtensionExecutor:
-    """docker 执行占位（Dev-B 的 extension_exec.py 替换实现）。
+    """适配 Dev-B 的 extension_exec.ExtensionExecutor（TL 接线 2026-09-12）。
 
-    本模块经 inject_executor(executor) 注入；路由层从 context 取。
-    占位实现 = 留痕执行（不走 docker，也不假装成功）。
+    对外仍是 ExtensionRunSpec/ExtensionRunResult 契约（契约 §3 服务端侧），
+    内部把 spec 翻译成 extension_exec.ExtensionRunRequest 后执行；
+    未注入真实执行器（inject_real_executor 未调用）时保持留痕 503 语义。
     """
 
+    def __init__(self) -> None:
+        self._real: Any | None = None
+
+    def inject_real(self, real: Any) -> None:
+        self._real = real
+
     async def execute(self, spec_in: ExtensionRunSpec) -> ExtensionRunResult:
-        raise ExtensionRunError(
-            "扩展执行器未接入（extension_exec.py 由 Dev-B 实现；"
-            "接口签名见 extensions.ExtensionExecutor）"
+        real = self._real
+        if real is None:
+            raise ExtensionRunError(
+                "扩展执行器未接入（extension_exec.py 由 Dev-B 实现；"
+                "接口签名见 extensions.ExtensionExecutor）"
+            )
+        from . import extension_exec as ex  # 延迟 import 避免循环
+        req = ex.ExtensionRunRequest(
+            image=str(spec_in.runtime_image),
+            ext_dir=str(spec_in.ext_dir),
+            input_dir=str(spec_in.input_dir),
+            output_dir=str(spec_in.output_dir),
+            params=dict(spec_in.params or {}),
+            cpu=spec_in.cpu,
+            memory=spec_in.memory,
         )
+        raw = await _call_executor(real, req)
+        if isinstance(raw, dict):
+            return _result_from_dict(raw, run_id=spec_in.log_path or new_run_id())
+        return raw
 
 
 # ---- 扩展生命周期状态机 ----
@@ -541,7 +564,17 @@ class ExtensionStore:
             fresh = self.load_extension(ext_id)
             fresh["validation"] = report
             self._set_status(fresh, STATUS_VALIDATED if ok else STATUS_TRIAL_FAILED)
-            return self.save_extension(fresh)
+            saved = self.save_extension(fresh)
+            if ok:
+                # TL 接线:Dev-C 发布卡依赖此帧(ready_to_publish=true 时同源挂发布卡)
+                self._publish("extension_validate", {
+                    "ext_id": ext_id,
+                    "validated": True,
+                    "ready_to_publish": True,
+                    "golden": golden,
+                    "form_defaults": self._confirm_form_defaults(saved),
+                })
+            return saved
 
         # 无金标：trial 通过即 validated
         trial = ext.get("trial") if isinstance(ext.get("trial"), dict) else {}
@@ -550,7 +583,32 @@ class ExtensionStore:
         fresh = self.load_extension(ext_id)
         fresh["validation"] = report
         self._set_status(fresh, STATUS_VALIDATED if ok else STATUS_TRIAL_FAILED)
-        return self.save_extension(fresh)
+        saved = self.save_extension(fresh)
+        if ok:
+            self._publish("extension_validate", {
+                "ext_id": ext_id,
+                "validated": True,
+                "ready_to_publish": True,
+                "golden": golden,
+                "form_defaults": self._confirm_form_defaults(saved),
+            })
+        return saved
+
+    def _confirm_form_defaults(self, ext: dict[str, Any]) -> list[dict[str, Any]]:
+        """confirm/publish 表单默认值(契约 §4 extension_validate.form_defaults):
+        参数 schema 的 name/type/required/description/default。"""
+        out: list[dict[str, Any]] = []
+        for p in ext.get("parameters") or []:
+            if not isinstance(p, dict):
+                continue
+            out.append({
+                "name": p.get("name"),
+                "type": p.get("type") or "text",
+                "required": bool(p.get("required")),
+                "description": p.get("description") or "",
+                "default": p.get("default"),
+            })
+        return out
 
     # ---- repair（POST /{id}/repair：LLM 修复，上限 max_repair_rounds）----
 
@@ -861,6 +919,14 @@ def get_store() -> ExtensionStore:
     return _store
 
 
+def inject_real_executor(real: Any, store: ExtensionStore | None = None) -> None:
+    """把 Dev-B extension_exec.ExtensionExecutor 注入扩展执行器（app 启动接线）。"""
+    target = store if store is not None else _active_store
+    if target is None:
+        return
+    target.executor.inject_real(real)
+
+
 def inject_executor(store: ExtensionStore | None = None) -> None:
     """Dev-B 替换点：extension_exec.py 实现后调用本函数注入真执行器。