From 279ad22a201acdbbfab520ea913ddcff27af98b9 Mon Sep 17 00:00:00 2001 From: DemonGatanjieu <207458273+DemonGatanjieu@users.noreply.github.com> Date: Tue, 27 Oct 2026 03:58:05 +0800 Subject: [PATCH] Honor completed task follow-ups and count only completed target retries as recovery --- CHANGELOG.md | 3 + api/ai_task_handoff.py | 48 ++++++++++--- api/ai_tasks.py | 33 +++++---- docs/architecture/backend.md | 9 +++ tests/test_ai_completion_edges.py | 108 ++++++++++++++++++++++++++++++ 5 files changed, 176 insertions(+), 25 deletions(-) create mode 100644 tests/test_ai_completion_edges.py diff --git a/CHANGELOG.md b/CHANGELOG.md index a57368a..9494ed6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,9 @@ ## v1.62.0 — unreleased +- 修复完成后的任务接续：AI 带着原任务 ID 查看结果、补面板，也会留在刚完成的任务里，不会另开空任务或把面板放进另一个任务。错误标记只在对应重试完成并成功后消除；等待审批、仍在运行、已取消，以及另一个节点或设置的成功都不算修好。 +- Fixed follow-ups carrying an explicit task ID: reads and panels immediately after completion stay with that task instead of creating an empty task or placing a panel in another one. Error recovery requires a completed successful retry of the recorded target; pending approval, running or cancelled attempts and success on another node or setting do not clear it. + - 参数名自动纠正现在也用于审批和任务记录：你批准后，AI 改用正确参数名重试不再重复问，换模型的结果卡片也不会漏掉。展台和底部提问条同步同一道题的草稿，刷新时保留原来的光标、选区和编辑位置。 - Corrected argument names now carry through approval and task recording: retrying with the proper name reuses an unused approval, and model changes keep their result cards. The stage and bottom question strip share each answer draft; refresh preserves the caret, selection and editing surface. diff --git a/api/ai_task_handoff.py b/api/ai_task_handoff.py index 8f5cb57..187a7d7 100644 --- a/api/ai_task_handoff.py +++ b/api/ai_task_handoff.py @@ -5,6 +5,8 @@ Called under ai_tasks' lock when a caller has no recent open task. Explicit task and the user's task waiting for the next AI take precedence in ai_tasks. """ +import json + REPLIES = {"user_message", "user_marks", "user_pictures", "user_toolbox"} REPLY_FRESH = 2 * 3600 AFTER_DONE = 5 * 60 # a look this soon after the AI said its task is done is part of that task, not a new one @@ -31,27 +33,57 @@ def latest_replied(tasks, client, now): return task -def just_done(task, tasks, waiting, client, now): +def just_done(task, tasks, waiting, client, now, explicit=False): """`task` (the app's last one) when it said it is done (report done) moments ago: a look the AI takes right after (checking its result) goes there, the task staying ended, and what it shows or asks the user about it (a before and after, "happy with it?") opens it again; no new task is made - for either. None when the user started a task for the next AI (`waiting`) or wrote to one.""" + for either. Automatic selection yields to a waiting task or unread reply; an explicit + task id keeps the same precedence as selecting an open task.""" if not task or not task.get("ended") or not 0 <= now - task["ended"] < AFTER_DONE: return None if not any(step.get("tool") == "report" and step.get("at") == task["ended"] for step in task["steps"]): return None # it ended by a pause or another task, not by the AI saying it is done - if (waiting and not waiting.get("ended")) or latest_replied(tasks, client, now): + if not explicit and ((waiting and not waiting.get("ended")) or latest_replied(tasks, client, now)): return None return task +RECOVERY_TARGETS = { + "set_model": ("node_id",), "set_prompt": ("node_id", "role"), "set_values": ("node_id",), + "connect_nodes": ("from_node_id", "to_node_id", "input"), "remove_nodes": ("node_ids",), + "run_workflow": ("workflow",), "open_workflow": ("workflow",), + "get_model": ("model_id",), "move_model": ("model_id",), "download_model": ("url",), + "get_image_info": ("image",), "zoom_image": ("image",), +} + + +def recovery_target(tool, arguments): + """Keep target identity, not prompt text or changed values, to distinguish independent retries.""" + target = {key: arguments[key] for key in RECOVERY_TARGETS.get(tool, ()) if arguments.get(key) is not None} + if tool == "set_prompt": + target["role"] = arguments.get("role") or "positive" + for key, value in target.items(): + if key.endswith("node_id"): + target[key] = str(value) + elif key == "node_ids" and isinstance(value, list): + target[key] = sorted(map(str, value)) + if tool == "set_values" and isinstance(arguments.get("values"), dict): + values = arguments["values"] + if arguments.get("node_id") is None and values and all(isinstance(value, dict) for value in values.values()): + target["settings"] = {str(node): sorted(settings) for node, settings in values.items()} + else: + target["settings"] = sorted(values) + return json.dumps(target, sort_keys=True, ensure_ascii=False) if target else None + + def unresolved(steps): - """Failed steps the AI did not get past: no later call of the same tool went through (a wrong - argument corrected on the next call, a run refused and then run, is no failure of the task).""" + """Only completed successful calls to the same recorded target resolve an earlier failure. + Old records without target metadata can only be compared at tool level with each other.""" later_ok, count = set(), 0 for step in reversed(steps): - if step.get("ok"): - later_ok.add(step.get("tool")) - elif step.get("tool") not in later_ok: + key = (step.get("tool"), step.get("recovery_target")) + if step.get("ok") and step.get("state") in (None, "done"): + later_ok.add(key) + elif not step.get("ok") and key not in later_ok: count += 1 return count diff --git a/api/ai_tasks.py b/api/ai_tasks.py index c3a9caf..99ef48f 100644 --- a/api/ai_tasks.py +++ b/api/ai_tasks.py @@ -36,7 +36,7 @@ import uuid import folder_paths from . import action_catalog, activity_log -from .ai_task_handoff import just_done, latest_replied, unresolved +from .ai_task_handoff import just_done, latest_replied, recovery_target, unresolved from .path_utils import atomic_write_json FORMAT = 1 @@ -265,13 +265,21 @@ def active_task_id(client, session="", task_id=None): return found if found in tasks else None -def current_task(client, now=None, session="", task_id=None, follow_up=False): +def current_task(client, now=None, session="", task_id=None, follow_up=False, read_only=False): """The task this app's calls go to now: an explicit task, one the user started for the next AI, else its open one, unless it paused too long. `follow_up`: a panel, question or note, which - right after the AI said its task is done is about that task (ai_task_handoff.just_done).""" + right after the AI said its task is done is about that task (ai_task_handoff.just_done). + `read_only` keeps such a task ended while a follow-up reopens it; both honor an explicit id.""" now = time.time() if now is None else now with _lock: tasks = _load() + task = tasks.get(task_id or _current.get(_owner_key(client, session))) + if (follow_up or read_only) and just_done(task, tasks, tasks.get(_waiting[0]), client, now, explicit=bool(task_id)): + _bind_owner(client, session, task["id"]) + if follow_up: + task["ended"] = None + _save(task) + return task picked = _claim_explicit(tasks, task_id, client, session) if picked: return picked @@ -285,10 +293,6 @@ def current_task(client, now=None, session="", task_id=None, follow_up=False): if task and not task.get("ended"): task["ended"] = task.get("updated") _save(task) - if follow_up and just_done(task, tasks, tasks.get(_waiting[0]), client, now): - task["ended"] = None - _save(task) - return task return _take_replied(tasks, client, now, session) or _new_task(client, now=now, session=session) @@ -413,9 +417,10 @@ def begin_step(client, tool, arguments, started=None, read_only=False, session=" args = arguments if isinstance(arguments, dict) else {} what = action_catalog.entry(tool, read_only) step = {"id": uuid.uuid4().hex[:10], "at": started, "ms": 0, "tool": tool, "kind": what.kind, "stage": what.stage, - "ok": True, "state": "running", "summary": summary_of(tool, args), "results": []} + "ok": True, "state": "running", "summary": summary_of(tool, args), "results": [], + "recovery_target": recovery_target(tool, args)} with _lock: - task = (read_only and not task_id and _just_done(client, session, started)) or current_task(client, started, session=session, task_id=task_id) + task = current_task(client, started, session=session, task_id=task_id, read_only=read_only) _add_step(task, step) _save(task) _notify(task["id"]) @@ -425,13 +430,6 @@ def begin_step(client, tool, arguments, started=None, read_only=False, session=" return None -def _just_done(client, session, now): - """This app's task it said is done moments ago, for a look right after (ai_task_handoff.just_done).""" - tasks = _load() - task = tasks.get(_current.get(_owner_key(client, session)) or (None if session else _current.get(client))) - return just_done(task, tasks, tasks.get(_waiting[0]), client, now) - - def finish_step(client, tool, arguments, result, started, read_only=False, step_ref=None, session="", task_id=None): """Completes an in-flight step (or creates one if step_ref is None) with its final outcome. Never raises.""" if not action_catalog.is_step(tool) or tool in OWN_STEPS: @@ -458,13 +456,14 @@ def finish_step(client, tool, arguments, result, started, read_only=False, step_ state = "error" if failed else "done" if gate.get("used") else "allowed" error = (step.get("error") or "你没有允许 / You said no") if failed else "" if task is None: - task = (read_only and not task_id and _just_done(client, session, started)) or current_task(client, started, session=session, task_id=task_id) + task = current_task(client, started, session=session, task_id=task_id, read_only=read_only) if step is None: step = {"id": uuid.uuid4().hex[:10], "at": started, "ms": ms, "tool": tool, "kind": kind, "stage": what.stage, "ok": not failed, "state": state, "summary": summary, "results": results} _add_step(task, step) else: step.update(ms=ms, ok=not failed, state=state, summary=summary, results=results) + step["recovery_target"] = recovery_target(tool, args) if error: step["error"] = error else: diff --git a/docs/architecture/backend.md b/docs/architecture/backend.md index 488419c..376608d 100644 --- a/docs/architecture/backend.md +++ b/docs/architecture/backend.md @@ -272,6 +272,15 @@ it ranks unread user messages, marks, materials and panel answers by the input's own timestamp, not the task's last background update. Only inputs less than two hours old qualify; future timestamps and automatic timeout choices do not. A task with an in-flight step is excluded from automatic takeover; explicit task selection is unchanged. +For five minutes after `report(done=True)`, reads stay in the finished task and panels/notes +reopen it. `ai_tasks.current_task` applies this to both implicit and explicit task ids, including +step recording without a start receipt. Automatic follow-ups yield to waiting tasks and unread +replies; an explicit id keeps precedence and does not consume another task waiting for an AI. +Task error counts only treat completed successful steps as recovery, never running, asking, +allowed or cancelled steps. New receipts keep a compact `recovery_target` (explicit node, +connection, workflow or resource, and setting names for `set_values`), so another target's +success does not hide the failure. Legacy receipts lacking this identity retain tool-level +matching with each other; no targets are guessed or migrated into historical records. `ai_answer_wait.collect` is shared by `get_answers` and question tools: after the bounded wait, typing/deadline checks, automatic choices and a final nonblocking collection share the task lock. A user answer that won the race is returned in that call, without a stale stop note. diff --git a/tests/test_ai_completion_edges.py b/tests/test_ai_completion_edges.py new file mode 100644 index 0000000..b7a6050 --- /dev/null +++ b/tests/test_ai_completion_edges.py @@ -0,0 +1,108 @@ +"""Completion follow-ups and error recovery use real task records in temporary storage.""" +import unittest + +from test_ai_tasks import isolate, ok, failed +from api import ai_panels, ai_tasks +from api.ai_task_handoff import AFTER_DONE + + +class CompletionEdges(unittest.TestCase): + def setUp(self): + isolate() + + def done(self, session="one"): + task = ai_tasks.begin_task("A", "Finished", now=100, session=session) + ai_tasks.report("A", "Done", done=True, now=110, session=session) + return task + + def test_explicit_read_after_done_stays_in_finished_task(self): + task = self.done() + ref = ai_tasks.begin_step("A", "get_situation", {}, 111, True, session="one", task_id=task["id"]) + ai_tasks.finish_step("A", "get_situation", {}, ok({}), 111, True, step_ref=ref) + self.assertEqual(ref[0], task["id"]) + self.assertEqual(task["ended"], 110) + self.assertEqual(len(ai_tasks.list_tasks()), 1) + + def test_explicit_panel_uses_named_finished_task_not_another_active_task(self): + finished = self.done() + active = ai_tasks.begin_task("A", "Other", now=112, session="one", new_task=True) + panel = ai_panels.show("A", "Result", [{"type": "text", "text": "Before / after"}], + now=113, session="one", task_id=finished["id"]) + self.assertEqual(ai_panels._find(panel["id"])[0]["id"], finished["id"]) + self.assertIsNone(finished["ended"]) + self.assertFalse(active["steps"]) + + def test_read_without_step_reference_keeps_explicit_finished_task(self): + task = self.done() + ai_tasks.record("A", "describe_canvas", {}, ok({}), 111, True, session="one", task_id=task["id"]) + self.assertEqual(len(ai_tasks.list_tasks()), 1) + self.assertEqual(task["steps"][-1]["tool"], "describe_canvas") + self.assertEqual(task["ended"], 110) + + def test_running_retry_does_not_hide_previous_failure(self): + ai_tasks.record("A", "set_model", {"node_id": 1, "model_id": "missing"}, failed("Missing model"), 100) + ref = ai_tasks.begin_step("A", "set_model", {"node_id": 1, "model_id": "found"}, 101) + task = ai_tasks.get_task(ref[0]) + self.assertEqual(ai_tasks.task_summary(task)["errors"], 1) + ai_tasks.finish_step("A", "set_model", {"node_id": 1, "model_id": "found"}, ok({}), 101, step_ref=ref) + self.assertEqual(ai_tasks.task_summary(ai_tasks.get_task(ref[0]))["errors"], 0) + + def test_waiting_for_approval_is_not_a_successful_retry(self): + ai_tasks.record("A", "set_prompt", {"prompt": "x"}, failed("Failed"), 100) + ai_tasks.record("A", "set_prompt", {"prompt": "x"}, ok({"waiting_for_user": True}), 101) + self.assertEqual(ai_tasks.list_tasks()[0]["errors"], 1) + + def test_success_on_a_different_node_does_not_resolve_a_failure(self): + ai_tasks.record("A", "set_model", {"node_id": 1, "model_id": "missing"}, failed("Missing model"), 100) + ai_tasks.record("A", "set_model", {"node_id": 2, "model_id": "found"}, ok({}), 101) + self.assertEqual(ai_tasks.list_tasks()[0]["errors"], 1) + + def test_cancelled_run_is_not_a_successful_retry(self): + task = {"id": "cancelled", "steps": [ + {"tool": "run_workflow", "kind": "run", "ok": False, "state": "error", "results": []}, + {"tool": "run_workflow", "kind": "run", "ok": True, "state": "cancelled", "results": []}, + ]} + self.assertEqual(ai_tasks.task_summary(task)["errors"], 1) + + def test_automatic_followup_yields_to_a_waiting_task(self): + finished = self.done() + waiting = ai_tasks.start_task("Next", now=111) + ref = ai_tasks.begin_step("A", "get_situation", {}, 112, True, session="one") + self.assertEqual(ref[0], waiting["id"]) + self.assertEqual(finished["ended"], 110) + + def test_explicit_followup_does_not_consume_an_unrelated_waiting_task(self): + finished = self.done() + waiting = ai_tasks.start_task("Next", now=111) + ref = ai_tasks.begin_step("A", "get_situation", {}, 112, True, session="one", task_id=finished["id"]) + self.assertEqual(ref[0], finished["id"]) + self.assertTrue(waiting["waiting"]) + self.assertEqual(ai_tasks._waiting[0], waiting["id"]) + + def test_automatic_followup_does_not_cross_sessions_or_completion_window(self): + for session, when in (("two", 111), ("one", 110 + AFTER_DONE)): + with self.subTest(session=session, when=when): + isolate() + finished = self.done() + ref = ai_tasks.begin_step("A", "get_situation", {}, when, True, session=session) + self.assertNotEqual(ref[0], finished["id"]) + + def test_a_different_setting_does_not_resolve_the_failed_setting(self): + ai_tasks.record("A", "set_values", {"node_id": 1, "values": {"steps": 0}}, failed("Invalid steps"), 100) + ai_tasks.record("A", "set_values", {"node_id": 1, "values": {"cfg": 5}}, ok({}), 101) + self.assertEqual(ai_tasks.list_tasks()[0]["errors"], 1) + ai_tasks.record("A", "set_values", {"node_id": "1", "values": {"steps": 20}}, ok({}), 102) + self.assertEqual(ai_tasks.list_tasks()[0]["errors"], 0) + + def test_recovery_target_survives_reloading_task_records(self): + ai_tasks.record("A", "set_model", {"node_id": 1, "model_id": "missing"}, failed("Missing"), 100) + ai_tasks.record("A", "set_model", {"node_id": 2, "model_id": "present"}, ok({}), 101) + ai_tasks._tasks = None + self.assertEqual(ai_tasks.list_tasks()[0]["errors"], 1) + + def test_batch_settings_keep_node_and_setting_identity(self): + ai_tasks.record("A", "set_values", {"values": {"1": {"steps": 0}}}, failed("Invalid steps"), 100) + ai_tasks.record("A", "set_values", {"values": {"1": {"cfg": 5}}}, ok({}), 101) + self.assertEqual(ai_tasks.list_tasks()[0]["errors"], 1) + ai_tasks.record("A", "set_values", {"values": {"1": {"steps": 20}}}, ok({}), 102) + self.assertEqual(ai_tasks.list_tasks()[0]["errors"], 0)