From 3b16397baf9d4b04e8da51fc6db237081630308f Mon Sep 17 00:00:00 2001 From: admin Date: Mon, 14 Sep 2026 13:53:44 +0800 Subject: [PATCH] =?UTF-8?q?feat(extension):=20TL=20=E6=8E=A5=E7=BA=BF?= =?UTF-8?q?=E2=80=94=E2=80=94executor=20=E9=80=82=E9=85=8D/validate=20?= =?UTF-8?q?=E5=B8=A7/form=5Fdefaults/context=20=E6=B3=A8=E5=85=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 --- bioflow_service/app/context.py | 13 +++ .../app/services/bioflow/extensions.py | 84 +++++++++++++++++-- 2 files changed, 88 insertions(+), 9 deletions(-) diff --git a/bioflow_service/app/context.py b/bioflow_service/app/context.py index 4301325..acef6fd 100644 --- a/bioflow_service/app/context.py +++ b/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 diff --git a/bioflow_service/app/services/bioflow/extensions.py b/bioflow_service/app/services/bioflow/extensions.py index 5ca6b91..95d8afd 100644 --- a/bioflow_service/app/services/bioflow/extensions.py +++ b/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 实现后调用本函数注入真执行器。