Fix job sheet stop races and early result receipts

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

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 两种都认不出。已经扫过的模型重新扫描后才会更新。

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 ""

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()