forked from admin/PlantAI-rag-new
fix(bioflow): 卡片防误操作+暂停立即生效+确认幂等
- 可调整参数卡「继续」成功后按钮隐藏,显示「参数已确认」 - 确认分析路径卡「确认创建任务」成功后按钮隐藏,显示「任务已创建」 - 后端 confirm_stage 幂等:已确认且 running/queued 的重复确认放行(不再 409) - 暂停立即生效:docker run 加 --name,docker kill 终止执行容器, running 阶段置 paused、任务即时暂停(kill 的 exit 137 不误判 failed) - 340 后端用例 + 121 r3debug + 373 主套件全绿;已部署 176 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
@@ -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)
|
||||
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -247,11 +247,13 @@ const continueText = computed(() => {
|
||||
>{{ autoContinue ? 'ON' : 'OFF' }}</button>
|
||||
</label>
|
||||
<button
|
||||
v-if="!props.block.submitted"
|
||||
class="ffc-continue"
|
||||
type="button"
|
||||
:disabled="submitting"
|
||||
@click="submit"
|
||||
>{{ continueText }}</button>
|
||||
<span v-else class="ffc-confirmed">{{ BF.formConfirmed }}</span>
|
||||
</div>
|
||||
<div v-if="submitError" class="ffc-submit-err">{{ submitError }}</div>
|
||||
</div>
|
||||
@@ -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;
|
||||
|
||||
@@ -165,13 +165,16 @@ defineExpose({ select: pick, confirm: confirmCreate });
|
||||
</div>
|
||||
</div>
|
||||
<div class="fpc-actions">
|
||||
<button class="pane-btn" type="button" :disabled="confirming" @click="backToSelect">{{ BF.jobsConfirmNo }}</button>
|
||||
<button
|
||||
class="fpc-confirm"
|
||||
type="button"
|
||||
:disabled="!canConfirm"
|
||||
@click="confirmCreate"
|
||||
>{{ confirmLabel }}</button>
|
||||
<template v-if="!props.block.jobId">
|
||||
<button class="pane-btn" type="button" :disabled="confirming" @click="backToSelect">{{ BF.jobsConfirmNo }}</button>
|
||||
<button
|
||||
class="fpc-confirm"
|
||||
type="button"
|
||||
:disabled="!canConfirm"
|
||||
@click="confirmCreate"
|
||||
>{{ confirmLabel }}</button>
|
||||
</template>
|
||||
<span v-else class="fpc-created">{{ BF.pathCreated }}</span>
|
||||
</div>
|
||||
<div v-if="error" class="fpc-error">{{ error }}</div>
|
||||
</template>
|
||||
@@ -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;
|
||||
|
||||
@@ -18,6 +18,8 @@ export const BF = {
|
||||
pathConfirm: '确认此路径',
|
||||
pathEmpty: '暂无候选路径',
|
||||
pathPlanning: '正在分析需求并规划分析路径…',
|
||||
formConfirmed: '参数已确认',
|
||||
pathCreated: '任务已创建',
|
||||
pathConfirming: '确认中…',
|
||||
pathFallback: '规划降级:',
|
||||
pathReuseNote: '复用存档',
|
||||
|
||||
Reference in New Issue
Block a user