From 20b3d6344e5cfe14df3e7b6bf76c75d19c310cc4 Mon Sep 17 00:00:00 2001 From: DemonGatanjieu <207458273+DemonGatanjieu@users.noreply.github.com> Date: Tue, 10 Nov 2026 21:02:29 +0800 Subject: [PATCH] Fix job sheet stop races and early result receipts --- ARCHITECTURE.md | 5 +- CHANGELOG.md | 3 ++ api/job_sheets.py | 31 +++++++----- tests/test_job_sheet_edges.py | 93 +++++++++++++++++++++++++++++++++++ 4 files changed, 118 insertions(+), 14 deletions(-) create mode 100644 tests/test_job_sheet_edges.py diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index c3c6165..9c9cfce 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -272,7 +272,10 @@ covered by the `styles.css` manifest. rules (`tests/job_sheet_plan.mjs`). `api/job_sheets.py` keeps the one sheet and where it stands, hears the page's reports, passes pause / stop / go on to it (from any page, the workbench's stop or the AI's stop_run; with its page gone another page takes it over) and writes a sheet the AI gave with `run_workflow(sheet=…)` on that call's run step, - so run_status and the workbench follow it like a run (`tests/test_job_sheets.py`). + so run_status and the workbench follow it like a run (`tests/test_job_sheets.py`). Clearing or replacing a stale + sheet stops its registered pending runs; a stop wins over a late pause report, and a queued run reported after + stopping is cancelled. Attaching the AI step carries over the sheet's recent pictures once. Interrupted and + removed runs count as unsuccessful (`tests/test_job_sheet_edges.py`). - A model's author text from Civitai in its details opens folded as the user last left it and can be translated into the window's language: `web/modules/detail_intro.js` (`tests/detail_intro.mjs`) keeps the fold and picks the language, `api/description_translation.py` translates it paragraph by paragraph through diff --git a/CHANGELOG.md b/CHANGELOG.md index 3ddaca6..ec9f797 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,9 @@ ## v1.64.0 — unreleased +- 任务单停止时，刚提交、迟到回报的运行也会取消，暂停回报不会覆盖停止请求；清除或替换失联任务单会同时停止它已登记的运行。很快完成、早于 AI 步骤绑定的图片会补回工作台，重复绑定不重复添图；中断或被移出队列的运行计入未成功次数。 +- Stopping a job sheet also cancels runs reported just after the stop; a pause report cannot erase a stop request. Clearing or replacing a stale sheet stops its registered runs. Pictures completed before the AI step was attached are carried into the workbench without duplicates on reattachment; interrupted or removed runs count as unsuccessful. + - 没有 C 站触发词的 LoRA（自己炼的、从别处下的），扫描时会从模型文件里读出训练图片大多都带的标签（作者写了触发词就用作者的），和 C 站的触发词分开存，搜模型时也能用这些词搜到。 - For a LoRA without trigger words from Civitai (trained yourself, downloaded elsewhere), a scan reads from the file the tags most of its training pictures share (or the trigger phrase its maker wrote) and keeps them apart from Civitai's trigger words; model search finds them too. - 从模型文件认底模时分得清 Flux.1 Dev 和 Schnell（Schnell 没有引导输入），也认得出 Z-Image 和 Lumina 2 这两种结构相近的模型：以前 Z-Image 的 GGUF 被当成 Lumina，safetensors 两种都认不出。已经扫过的模型重新扫描后才会更新。 diff --git a/api/job_sheets.py b/api/job_sheets.py index f4836d8..e8a8a51 100644 --- a/api/job_sheets.py +++ b/api/job_sheets.py @@ -150,7 +150,8 @@ def workflow(): def _end_old(sheet): """An unfinished sheet a new one replaces: it ends stopped, on its step too.""" if sheet and sheet["state"] not in ENDED: - sheet.update(state="stopped", want=None, note=sheet.get("note") or "replaced") + _drop(sheet["pending"]) + sheet.update(state="stopped", want=None, ended=time.time(), note=sheet.get("note") or "replaced") _sync_step(sheet) @@ -213,7 +214,7 @@ def _drop(prompt_ids): def report(body): """Word from the page running the sheet: {id, page, at, queued: [[prompt id, run]], finished: [[prompt id, - outcome]], state?, note?}; outcome "ok", "error" or "gone" (taken out of the queue). Answers the sheet, + outcome]], state?, note?}; outcome "ok", "error", "interrupted" or "gone" (taken out of the queue). Answers the sheet, whose `want` says what was asked of the page.""" pictures, dropped = [], [] with _lock: @@ -222,24 +223,28 @@ def report(body): raise SheetError("This sheet is gone.") if sheet["page"] != body.get("page"): raise SheetError("Another page runs this sheet now.") - if sheet["state"] in ENDED: - return public(sheet) - sheet["heard"] = time.time() + ended = sheet["state"] in ENDED known = {pid for pid, _ in sheet["ids"]} for item in body.get("queued") or []: if (isinstance(item, list) and len(item) == 2 and isinstance(item[0], str) and 0 < len(item[0]) <= 128 and isinstance(item[1], int) and item[0] not in known): - sheet["ids"].append(item) - sheet["pending"].append(item[0]) + if not ended: + sheet["ids"].append(item) + sheet["pending"].append(item[0]) known.add(item[0]) - if sheet["want"] == "stop": + if sheet["want"] == "stop" or sheet["state"] == "stopped": dropped.append(item[0]) # queued as stop was asked: it goes too + if ended: + _drop(dropped) # queuePrompt can return after the page's own Stop has already ended the sheet + return public(sheet) + sheet["heard"] = time.time() for item in body.get("finished") or []: - if not (isinstance(item, list) and len(item) == 2 and item[0] in sheet["pending"]): + if not (isinstance(item, list) and len(item) == 2 and item[0] in sheet["pending"] + and item[1] in ("ok", "error", "interrupted", "gone")): continue sheet["pending"].remove(item[0]) sheet["done"] += 1 - if item[1] == "error": + if item[1] != "ok": sheet["failed"] += 1 made = _pictures(item[0]) if item[1] == "ok" else [] pictures += made @@ -249,7 +254,7 @@ def report(body): at = body.get("at") if isinstance(at, int) and 0 <= at <= sheet["total"]: sheet["at"] = at - state = body.get("state") + state = "stopped" if sheet["want"] == "stop" else body.get("state") if state in ("running", "paused", "stopped", "done"): if state != sheet["state"]: sheet["want"] = None @@ -330,7 +335,7 @@ def attach(sheet_id, step_id): return sheet.update(task=task_id, step=step_id) atomic_write_json(_path(SHEET), sheet) - _sync_step(sheet) + _sync_step(sheet, sheet["last"]) def _words(sheet): @@ -354,7 +359,7 @@ def _sync_step(sheet, pictures=()): if pictures: kept = [item for item in step["results"] if item["type"] != "image"] shown = [item for item in step["results"] if item["type"] == "image"] - step["results"] = kept + (shown + list(pictures))[-ai_tasks.MAX_RESULTS:] + step["results"] = kept + (shown + [item for item in pictures if item not in shown])[-ai_tasks.MAX_RESULTS:] if sheet["state"] not in ENDED: step["state"] = "running" paused = " · 暂停 / paused" if sheet["state"] == "paused" else "" diff --git a/tests/test_job_sheet_edges.py b/tests/test_job_sheet_edges.py new file mode 100644 index 0000000..4a7caf2 --- /dev/null +++ b/tests/test_job_sheet_edges.py @@ -0,0 +1,93 @@ +"""Job-sheet cancellation and receipts when the page and MCP return in a different order.""" +import sys +import time +import unittest +from pathlib import Path +from unittest import mock + +PLUGIN = Path(__file__).resolve().parents[1] +sys.path[:0] = [str(PLUGIN), str(PLUGIN.parents[1]), str(PLUGIN / "tests")] + +from api import ai_tasks, job_sheets # noqa: E402 +from test_job_sheets import FLOW, OTHER, PAGE, Queue, isolate, picture # noqa: E402 + + +class SheetEdgeTests(unittest.TestCase): + def setUp(self): + isolate() + self.queue = Queue() + patch = mock.patch("api.ai_task_runs._queue", return_value=self.queue) + patch.start() + self.addCleanup(patch.stop) + + def start(self, **given): + return job_sheets.start({"rows": ["a cat"], "workflow": FLOW, "page": PAGE, **given}) + + def report(self, sheet, **fields): + return job_sheets.report({"id": sheet["id"], "page": PAGE, **fields}) + + def step(self, task): + result = {"structuredContent": {}, "content": [{"type": "text", "text": "ok"}]} + return ai_tasks.finish_step("AI", "run_workflow", {"sheet": {}}, result, time.time(), task_id=task) + + def test_clearing_a_stale_sheet_stops_its_pending_runs(self): + sheet = self.start() + self.report(sheet, queued=[["old-run", 0]], at=1) + with mock.patch.object(job_sheets.time, "time", return_value=time.time() + job_sheets.STALE + 1): + job_sheets.clear(sheet["id"]) + self.assertEqual(self.queue.interrupted, ["old-run"]) + self.assertTrue(self.queue.deleted[0]((0, "old-run"))) + self.assertFalse(self.queue.deleted[0]((0, "someone-else"))) + self.assertIsNone(job_sheets.current()) + + def test_replacing_a_stale_sheet_stops_only_its_runs(self): + sheet = self.start() + self.report(sheet, queued=[["old-run", 0]], at=1) + with mock.patch.object(job_sheets.time, "time", return_value=time.time() + job_sheets.STALE + 1): + new = self.start(page=OTHER) + self.assertEqual(self.queue.interrupted, ["old-run"]) + self.assertEqual(job_sheets.current()["id"], new["id"]) + self.assertNotEqual(new["id"], sheet["id"]) + + def test_a_late_queue_receipt_after_stop_is_cancelled(self): + sheet = self.start() + job_sheets.ask(sheet["id"], "stop", PAGE) + now = self.report(sheet, queued=[["late-run", 0]], at=1, state="running") + self.assertEqual(self.queue.interrupted, ["late-run"]) + self.assertEqual(now["state"], "stopped") + self.assertEqual(job_sheets.current()["state"], "stopped") + + def test_a_report_of_pause_does_not_erase_a_stop_request(self): + sheet = self.start() + job_sheets.ask(sheet["id"], "stop", OTHER) + now = self.report(sheet, state="paused") + self.assertTrue(now["state"] == "stopped" or now["want"] == "stop") + + def test_binding_the_ai_step_keeps_earlier_pictures_once(self): + task = ai_tasks.start_task("fast sheet")["id"] + sheet = self.start(task=task) + self.queue.history["fast-run"] = picture("cached.png") + self.report(sheet, queued=[["fast-run", 0]], finished=[["fast-run", "ok"]], at=1, state="done") + step = self.step(task) + job_sheets.attach(sheet["id"], step["id"]) + job_sheets.attach(sheet["id"], step["id"]) + shown = next(item for item in ai_tasks.get_task(task)["steps"] if item["id"] == step["id"]) + self.assertEqual([item["filename"] for item in shown["results"] if item["type"] == "image"], ["cached.png"]) + self.assertEqual(shown["state"], "done") + + def test_non_successful_runs_do_not_become_a_successful_sheet(self): + for outcome in ("interrupted", "gone"): + with self.subTest(outcome=outcome): + isolate() + task = ai_tasks.start_task("interrupted sheet")["id"] + sheet = self.start(task=task) + step = self.step(task) + job_sheets.attach(sheet["id"], step["id"]) + self.report(sheet, queued=[["p1", 0]], finished=[["p1", outcome]], at=1, state="done") + shown = next(item for item in ai_tasks.get_task(task)["steps"] if item["id"] == step["id"]) + self.assertEqual(shown["state"], "error") + self.assertFalse(shown["ok"]) + + +if __name__ == "__main__": + unittest.main()