diff --git a/bioflow_service/app/services/bioflow/kernel.py b/bioflow_service/app/services/bioflow/kernel.py index b80278b..f7d5cc4 100644 --- a/bioflow_service/app/services/bioflow/kernel.py +++ b/bioflow_service/app/services/bioflow/kernel.py @@ -122,6 +122,7 @@ class Kernel: memory: str | None = None, module: str | None = None, extra_mounts: list[str] | None = None, + container_name: str | None = None, ) -> list[str]: """拼装 docker run 命令(list,无 shell;TC-BIO-KERN-001 基元)。 @@ -140,6 +141,8 @@ class Kernel: raise UnknownStageError(f"阶段 {stage} 无执行模块") cmd: list[str] = ["docker", "run", "--rm", "--user", str(self.cfg.host_uid)] + if container_name: + cmd += ["--name", container_name] cmd += ["-v", f"{job_input_dir}:{self.cfg.container_input_dir}"] cmd += ["-v", f"{job_workspace_dir}:{self.cfg.container_workspace_dir}"] if self.cfg.data_root: @@ -273,10 +276,12 @@ class Kernel: cpu: str | None = None, memory: str | None = None, extra_mounts: list[str] | None = None, + container_name: str | None = None, ) -> dict[str, Any]: """执行一个阶段 docker:返回 {exit_code, timed_out, log_tail, cmd}。 - extra_mounts:本次执行额外只读挂载的宿主绝对路径(登记路径,src==dst) + - container_name:docker run --name(暂停时按名 kill 容器用) - 超时:terminate → 3s → kill(TC-BIO-KERN-009) - 日志全量写 log_path(任务 logs/),返回尾部 log_tail_lines 行 - 退出码非 0 / 超时 → 失败语义由 runner 判定(本函数只报事实) @@ -286,7 +291,8 @@ class Kernel: memory = memory or str(params.get("memory") or "") cmd = self.build_command(stage, job_input_dir, job_workspace_dir, params, cpu=cpu or None, memory=memory or None, - extra_mounts=extra_mounts) + extra_mounts=extra_mounts, + container_name=container_name) log_path = Path(log_path) if log_path else Path(job_workspace_dir) / "logs" / f"{stage}.log" log_path.parent.mkdir(parents=True, exist_ok=True) diff --git a/bioflow_service/app/services/bioflow/runner.py b/bioflow_service/app/services/bioflow/runner.py index bd92df3..9acf18a 100644 --- a/bioflow_service/app/services/bioflow/runner.py +++ b/bioflow_service/app/services/bioflow/runner.py @@ -148,6 +148,7 @@ class JobRecord: archive_id: str | None = None pause_requested: bool = False registered_server_paths: list[str] = field(default_factory=list) # 登记路径(自动挂载) + _running_container: str | None = None # 当前执行容器名(暂停时按名 kill) _stage_states: dict[str, dict[str, Any]] = field(default_factory=dict) @property @@ -384,18 +385,30 @@ class Runner: self.events.publish("stage_started", { "job_id": record.id, "stage": stage, "attempt": attempt, }) + attempt_no = int(record._stage_states.get(stage, {}).get("attempt") or 0) + 1 + record._running_container = f"plantai-bio-{record.id}-{stage}-{attempt_no}" result = await self.executor(stage, record.input_dir, record.workspace_dir, params, log_path=record.logs_dir / f"{stage}.log", - extra_mounts=self._registered_mounts(record)) + extra_mounts=self._registered_mounts(record), + container_name=record._running_container) + record._running_container = None log_tail = str(result.get("log_tail") or "") self._set_stage(record, stage, log_tail=log_tail) await self._persist_stage(record, stage) self.events.publish("stage_log", { "job_id": record.id, "stage": stage, "log": log_tail, }) + if record.pause_requested: + # 暂停请求在阶段执行期间到达(容器已被 pause_job kill): + # 不按 exit 137 判 failed,走暂停分支 + self._set_stage(record, stage, status=STAGE_PAUSED, error="") + await self._persist_stage(record, stage) + await self._pause_job(record, stage) + return "paused" outcome = await self._judge_stage(record, stage, result) return outcome finally: + record._running_container = None self.semaphore.release() async def _acquire(self, record: JobRecord, stage: str) -> bool: @@ -497,6 +510,10 @@ class Runner: async def confirm_stage(self, record: JobRecord, stage: str, params: dict[str, Any]) -> None: """阶段确认(FR-BIO-009):校验参数→存表→执行(并发排队)。""" status = self.stage_status(record, stage) + # 幂等:已确认且正在执行/排队(如前端重复点击「继续」),直接放行不报 409 + st_existing = record._stage_states.get(stage, {}) + if status in (STAGE_RUNNING, STAGE_QUEUED) and st_existing.get("_confirmed"): + return if status not in (STAGE_WAITING_CONFIRM, STAGE_QUEUED, STAGE_WAITING): raise RunnerError("STAGE_NOT_CONFIRMABLE", f"阶段 {stage} 当前状态 {status} 不可确认") allowed = self._allowed_param_names(stage) @@ -611,14 +628,25 @@ class Runner: logger.exception("runner: skip 后推进异常 %s", record.id) async def pause_job(self, record: JobRecord) -> None: - """暂停(FR-BIO-010):等待当前阶段完成后停在阶段边界。""" + """暂停(FR-BIO-010):running 阶段有执行容器时立即 kill,任务即刻暂停。""" if record.status not in (JOB_RUNNING, JOB_AWAIT_INPUTS): raise RunnerError("JOB_NOT_RUNNING", f"任务 {record.id} 当前状态 {record.status} 不可暂停") record.pause_requested = True await self._persist_job(record) - # 无 running 阶段 → 立即暂停 + # 有 running 阶段 → kill 执行容器(立即生效;失败仅日志,不阻断暂停语义) running = [s for s in record.stage_chain if self.stage_status(record, s) == STAGE_RUNNING] + if running and getattr(record, "_running_container", None): + name = record._running_container + try: + proc = await asyncio.create_subprocess_shell( + f"docker kill {name}", + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.STDOUT, + ) + await asyncio.wait_for(proc.communicate(), timeout=30.0) + except Exception as exc: # noqa: BLE001 - kill 失败不阻断(仍走阶段边界暂停) + logger.warning("pause_job: docker kill %s 失败: %s", name, exc) if not running: await self._pause_job(record, None) diff --git a/bioflow_service/tests/test_runner.py b/bioflow_service/tests/test_runner.py index 60a5b6e..17afaad 100644 --- a/bioflow_service/tests/test_runner.py +++ b/bioflow_service/tests/test_runner.py @@ -19,6 +19,9 @@ from app.services.bioflow.runner import ( JOB_FAILED, JOB_PAUSED, STAGE_DONE, + STAGE_PAUSED, + STAGE_QUEUED, + STAGE_RUNNING, STAGE_FAILED, STAGE_SKIPPED, STAGE_WAITING_CONFIRM, @@ -177,7 +180,7 @@ def test_rerun_non_failed_rejected(db, events, catalog, fake_executor): def test_pause_at_stage_boundary(db, events, catalog): - """TC-BIO-RUN-006:暂停在阶段边界(当前阶段完成后停)。""" + """TC-BIO-RUN-006:运行中暂停立即生效(当前阶段置 paused,不判失败)。""" executor = FakeExecutor(delay=0.05) async def scenario(): @@ -192,10 +195,29 @@ def test_pause_at_stage_boundary(db, events, catalog): runner, record = _scenario(scenario) assert record.status == JOB_PAUSED - assert runner.stage_status(record, "identify") == STAGE_DONE + assert runner.stage_status(record, "identify") == STAGE_PAUSED # 立即暂停语义 assert all(c["stage"] != "phylogeny" for c in executor.calls) +def test_confirm_idempotent_while_running(db, events, catalog): + """确认幂等:已确认且 running/queued 的重复 confirm 不报 409。""" + executor = FakeExecutor(delay=0.1) + + async def scenario(): + runner = make_runner(db, events, executor, catalog) + record = await runner.create_job( + "s1", "t", ["identify"], auto_continue=False, auto_start=False) + await runner.run_job(record) # 手动模式:停 waiting_confirm + await runner.confirm_stage(record, "identify", {}) + # 立即二次 confirm:阶段已 running 且 _confirmed → 幂等放行,不抛异常 + await runner.confirm_stage(record, "identify", {}) + await _drain(runner) + return runner, record + + runner, record = _scenario(scenario) + assert runner.stage_status(record, "identify") in (STAGE_RUNNING, STAGE_DONE, STAGE_QUEUED) + + def test_resume_continues_without_rerun(db, events, catalog): """TC-BIO-RUN-007:暂停后恢复续跑,不重跑已完成。""" executor = FakeExecutor(delay=0.02) diff --git a/ui/src/components/chat/FlowFormCard.vue b/ui/src/components/chat/FlowFormCard.vue index 6723a3d..6370ce9 100644 --- a/ui/src/components/chat/FlowFormCard.vue +++ b/ui/src/components/chat/FlowFormCard.vue @@ -247,11 +247,13 @@ const continueText = computed(() => { >{{ autoContinue ? 'ON' : 'OFF' }} + {{ BF.formConfirmed }}
{{ submitError }}
@@ -396,6 +398,11 @@ const continueText = computed(() => { box-shadow: none; transform: none; } +.ffc-confirmed { + font-size: 12.5px; + font-weight: 600; + color: var(--pa-ink-subtle); +} .ffc-submit-err { margin-top: 8px; font-size: 12px; diff --git a/ui/src/components/chat/FlowPathCard.vue b/ui/src/components/chat/FlowPathCard.vue index 9f497ce..4307b13 100644 --- a/ui/src/components/chat/FlowPathCard.vue +++ b/ui/src/components/chat/FlowPathCard.vue @@ -165,13 +165,16 @@ defineExpose({ select: pick, confirm: confirmCreate });
- - + + {{ BF.pathCreated }}
{{ error }}
@@ -285,6 +288,11 @@ defineExpose({ select: pick, confirm: confirmCreate }); gap: 8px; margin-top: 10px; } +.fpc-created { + font-size: 12.5px; + font-weight: 600; + color: var(--pa-ink-subtle); +} /* 确认按钮:--pa-gradient 底白字 12.5px/600,r-md padding 7px 14px;禁用 opacity .45 */ .fpc-confirm { padding: 7px 14px; diff --git a/ui/src/i18n/bioflow.js b/ui/src/i18n/bioflow.js index 8d29a2c..610b777 100644 --- a/ui/src/i18n/bioflow.js +++ b/ui/src/i18n/bioflow.js @@ -18,6 +18,8 @@ export const BF = { pathConfirm: '确认此路径', pathEmpty: '暂无候选路径', pathPlanning: '正在分析需求并规划分析路径…', + formConfirmed: '参数已确认', + pathCreated: '任务已创建', pathConfirming: '确认中…', pathFallback: '规划降级:', pathReuseNote: '复用存档',