diff --git a/CHANGELOG.md b/CHANGELOG.md index 83efea8..df8990d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,9 @@ ## v1.62.0 — unreleased +- 修复任务接续与等待回答：按你最近一次未读回复接续任务，后台进度更新不再打乱顺序；选择题回答也能接回原任务，仍有工具正在执行的任务不会被新会话自动接走。读取答案不指定面板时也会正确处理超时；到时间自动选择或恰好收到你的回答，都会在当次调用里交给 AI。等选择题时发的留言和图片不再被读走却漏传。 +- Fixed task resumption and answer waiting: resume by the latest unread user reply, including panel answers, rather than background updates; a task with a tool still running is excluded from automatic takeover. Reading answers without a panel now handles expired questions too. Timeout choices and user answers arriving at that boundary reach the AI in the same call; messages and pictures sent while it waits for a question are no longer consumed without delivery. + - AI 少走弯路、少猜： - 去掉了“给 AI 的工具”里的 **日常**：所有工具都直接给。Antigravity、Claude Code、Cursor、VS Code 这些软件会自己按需读工具说明，AMB 再做一层“现查现用”只会让 AI 多绕路（一次实测 29 次调用里 18 次花在查工具和转一道上）。 - AMB 更新以后，AI 软件还拿着旧的工具列表来调用时，任务标题下面会出现一行黄字，告诉你怎么让它重新连接；AI 那边也收到同样的提醒。AI 软件重新取了列表，黄字自己消失。 diff --git a/api/ai_answer_wait.py b/api/ai_answer_wait.py index 58595e8..419081a 100644 --- a/api/ai_answer_wait.py +++ b/api/ai_answer_wait.py @@ -38,23 +38,29 @@ def take_recommended(task_id, step): def typing(task_id, now=None): """The user is typing in task `task_id` (an answer or a message). False for no such task.""" - if not ai_tasks.get_task(task_id): - return False - _typed[task_id] = time.time() if now is None else now - return True + with ai_tasks._lock: + if not ai_tasks.get_task(task_id): + return False + _typed[task_id] = time.time() if now is None else now + return True -def wait_on(panel_id, now=None): +def wait_on(panel_id, now=None, allow_recommended=True): """For a question of panel `panel_id` still open: {keep_waiting, seconds_left?, typing?, note}, or {}.""" with ai_tasks._lock: task, step = ai_panels._find(panel_id) + return _wait_on(task, step, now, allow_recommended) + + +def _wait_on(task, step, now, allow_recommended): + """Check typing and commit a timeout choice under the same lock as user answers.""" if not step: return {} now = time.time() if now is None else now typed = _typed.get(task["id"], 0) left = max(step["at"] + ai_gates.answer_wait(), typed + AFTER_TYPING) - now if left <= 0: - if ai_gates.no_answer() == "recommended" and take_recommended(task["id"], step): + if allow_recommended and ai_gates.no_answer() == "recommended" and take_recommended(task["id"], step): return {"keep_waiting": False, "took_recommended": True, "note": TOOK} return {"keep_waiting": False, "note": "You waited as long as the user set: end your turn now. They answer in " "Anomalous and tell you in your chat to go on."} @@ -72,6 +78,28 @@ def for_answer(answer, panel_id=None, now=None): if not waiting or answer.get("answers") or answer.get("messages"): return answer if panel_id is None: - found = [found for item in waiting if (found := wait_on(item["panel"], now)).get("keep_waiting")] - return {**answer, **max(found, key=lambda item: item["seconds_left"])} if found else answer + found = [wait_on(panel, now) for panel in dict.fromkeys(item["panel"] for item in waiting)] + live = [item for item in found if item.get("keep_waiting")] + status = max(live, key=lambda item: item["seconds_left"]) if live else next((item for item in found if item), {}) + took = any(item.get("took_recommended") for item in found) + return {**answer, **status, **({"took_recommended": True} if took else {})} return {**answer, **wait_on(panel_id, now)} + + +def collect(client, panel_id=None, wait_seconds=0, task_id=None): + """Wait unlocked, then deliver a timeout choice or a racing user answer in this same call.""" + answer = ai_panels.get_answers(client, panel_id, wait_seconds, task_id=task_id) + if answer.get("answers") or answer.get("messages") or not answer.get("still_waiting"): + return answer + with ai_tasks._lock: + # Input may have arrived after the long wait released its lock. Read it before choosing. + answer = ai_panels.get_answers(client, panel_id, 0, task_id=task_id) + if answer.get("answers") or answer.get("messages") or not answer.get("still_waiting"): + return answer + status = for_answer(answer, panel_id) + fresh = ai_panels.get_answers(client, panel_id, 0, task_id=task_id) + if fresh.get("answers") or fresh.get("messages"): + if any(item.get("recommended_on_timeout") for item in fresh.get("answers", [])): + fresh["note"] = TOOK + return fresh + return {**status, **fresh} if fresh.get("still_waiting") else fresh diff --git a/api/ai_task_handoff.py b/api/ai_task_handoff.py new file mode 100644 index 0000000..798bd16 --- /dev/null +++ b/api/ai_task_handoff.py @@ -0,0 +1,29 @@ +"""Select a task to resume from unread user input, without changing task state or ownership. + +Called under ai_tasks' lock when a caller has no recent open task. Explicit task selection +and the user's task waiting for the next AI take precedence in ai_tasks. +""" + +REPLIES = {"user_message", "user_marks", "user_pictures", "user_toolbox"} +REPLY_FRESH = 2 * 3600 + + +def _reply_at(task, now): + """Latest unread user input; a tool still running keeps its task from automatic takeover.""" + if any(step.get("state") == "running" for step in task["steps"]): + return 0 + replies = [] + for step in task["steps"]: + if step.get("kind") == "user" and step.get("tool") in REPLIES: + replies.append(step) + elif step.get("kind") == "panel": + replies.extend((step.get("answers") or {}).values()) + return max((r.get("at", 0) for r in replies if not r.get("delivered") and not r.get("recommended_on_timeout") + and 0 <= now - r.get("at", 0) < REPLY_FRESH), default=0) + + +def latest_replied(tasks, client, now): + """The task of this app with the latest eligible reply, or None.""" + candidates = ((at, task) for task in tasks.values() if task.get("client") == client and (at := _reply_at(task, now))) + _, task = max(candidates, key=lambda item: item[0], default=(0, None)) + return task diff --git a/api/ai_tasks.py b/api/ai_tasks.py index adee469..a8c0fed 100644 --- a/api/ai_tasks.py +++ b/api/ai_tasks.py @@ -36,6 +36,7 @@ import uuid import folder_paths from . import action_catalog, activity_log +from .ai_task_handoff import latest_replied from .path_utils import atomic_write_json FORMAT = 1 @@ -287,17 +288,10 @@ def current_task(client, now=None, session="", task_id=None): return _take_replied(tasks, client, now, session) or _new_task(client, now=now, session=session) -REPLIES = {"user_message", "user_marks", "user_pictures", "user_toolbox"} # what the user gives the AI (ai_call_lines.js) -REPLY_FRESH = 2 * 3600 # an unread reply older than this no longer draws a new chat's calls to its task - - def _take_replied(tasks, client, now, session="", previous=None): """The newest task of this app the user wrote to lately, since its AI last read (an ended one opens again), now this app's instead of a new one (its own open task `previous` ends), or None.""" - replied = max((task for task in tasks.values() if task.get("client") == client and any( - step.get("kind") == "user" and step.get("tool") in REPLIES and not step.get("delivered") - and now - step.get("at", 0) < REPLY_FRESH for step in task["steps"])), - key=lambda task: task.get("updated", 0), default=None) + replied = latest_replied(tasks, client, now) if replied is None: return None if previous and previous is not replied and not previous.get("ended"): diff --git a/api/mcp_task_tools.py b/api/mcp_task_tools.py index 3a18c6b..c9a3ba3 100644 --- a/api/mcp_task_tools.py +++ b/api/mcp_task_tools.py @@ -134,16 +134,18 @@ def _answer_of(ctx, panel_id, wait_seconds): """The answer to the question of panel `panel_id` once the user gives it, waiting at least ASK_WAIT_MIN.""" client = ctx.get("client") or "AI" try: - answer = ai_panels.get_answers(client, panel_id, max(ASK_WAIT_MIN, float(wait_seconds or 0))) + answer = ai_answer_wait.collect(client, panel_id, max(ASK_WAIT_MIN, float(wait_seconds or 0))) given = next((item for item in answer["answers"] if item["question"] == "answer"), None) if not given: - waited = ai_answer_wait.wait_on(panel_id) - if not waited.get("took_recommended"): - return {"answered": False, "panel": panel_id, **waited} - given = next((item for item in ai_panels.get_answers(client, panel_id, 0)["answers"] if item["question"] == "answer"), {}) + if answer.get("messages") and answer.get("still_waiting"): + # Read the user's new instruction before considering an automatic choice. + answer.update(ai_answer_wait.wait_on(panel_id, allow_recommended=False)) + return {"answered": False, "panel": panel_id, **answer} except ai_panels.PanelError as error: raise ToolError(str(error)) out = {"answered": True, "panel": panel_id, "next": "Go on with the user's answer; log_decision it when it settles a choice."} + if answer.get("messages"): + out["messages"] = answer["messages"] if given.get("inpaint_image"): out.update(inpaint_image=given["inpaint_image"], inpaint_note=given["note"]) elif given.get("note"): @@ -248,9 +250,7 @@ def get_answers(ctx, panel=None, wait_seconds=0, task=None): task_id = task or ctx.get("task") or None if not task_id and ctx.get("session"): task_id = ai_tasks.active_task_id(client, ctx["session"]) or "" - answer = ai_answer_wait.for_answer(ai_panels.get_answers(client, panel, wait_seconds, task_id=task_id), panel) - if answer.get("took_recommended"): # Anomalous answered for the user just now: it goes in this answer - answer = {**ai_panels.get_answers(client, panel, 0, task_id=task_id), "note": ai_answer_wait.TOOK} + answer = ai_answer_wait.collect(client, panel, wait_seconds, task_id=task_id) return answer, ai_marks.image_items(answer.get("messages") or []) # marked pictures, as pictures except ai_panels.PanelError as error: raise ToolError(str(error)) @@ -288,6 +288,14 @@ def _task_tool(name, title, description, run, properties, required=()): }} +def _question_with_images(run): + """The question receipt also carries pictures the user sent while it was waiting.""" + def call(ctx, **arguments): + result = run(ctx, **arguments) + return result, ai_marks.image_items(result.get("messages") or []) + return call + + TASK_TOOLS = dict([ _task_tool("track_run", "Track an externally submitted run", "Immediately after YOU submit to this ComfyUI server via /prompt or another tool, attach the exact " @@ -345,7 +353,7 @@ TASK_TOOLS = dict([ "other=false; no options: a free answer. Use it for every choice that is the user's, not your chat's " f"question form. Waits at least {ASK_WAIT_MIN} s, up to wait_seconds; unanswered and keep_waiting: wait on " "with get_answers. End every round of work with it.", - ask_user, {"question": {"type": "string"}, + _question_with_images(ask_user), {"question": {"type": "string"}, "options": {"type": "array", "maxItems": 8, "items": {"type": "object", "properties": { "label": {"type": "string"}, "detail": {"type": "string"}, "recommended": {"type": "boolean"}}}}, "multiple": {"type": "boolean", "default": False}, "other": {"type": "boolean", "default": True}, @@ -359,7 +367,7 @@ TASK_TOOLS = dict([ "soft edge. The user sees theirs and yours side by side and picks one, or marks again; with an area " "picked the answer has inpaint_image (the picture with that area as its alpha, in ComfyUI's input " "folder) and inpaint_note. Build the repaint yourself. Waits like ask_user.", - propose_repaint_area, {"title": {"type": "string", "description": "A few words, in the user's language."}, + _question_with_images(propose_repaint_area), {"title": {"type": "string", "description": "A few words, in the user's language."}, "marks": {"type": "string"}, "areas": {"type": "array", "maxItems": ai_repaint.MAX_AREAS, "items": {"type": "object", "properties": { "shape": {"type": "string", "enum": list(ai_repaint.SHAPES)}, "x": {"type": "number"}, diff --git a/docs/architecture/backend.md b/docs/architecture/backend.md index d1c1369..f3328ea 100644 --- a/docs/architecture/backend.md +++ b/docs/architecture/backend.md @@ -267,6 +267,17 @@ delivery. Gate decisions wake both gate waiters and `get_answers` readers. Expli selection in `ai_panels._collect` is exclusive, including when that task no longer exists; session-scoped `get_answers` resolves its own active task before collecting input. +`api/ai_task_handoff.py` selects automatic resumption candidates without mutating ownership: +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. +`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. +Without a panel id, deadlines are checked once per waiting panel, including expired ones. +Question receipts preserve messages and attach their pictures even when no option was chosen. + Run monitoring failures end the receipt as `stopped` with a reason without cancelling the host workflow. No detected queue entry is not success. Batch progress keeps the complete prompt list, and partial errors are saved even before all prompts finish. Marked-image diff --git a/tests/test_ai_handoff_edges.py b/tests/test_ai_handoff_edges.py new file mode 100644 index 0000000..f5f885b --- /dev/null +++ b/tests/test_ai_handoff_edges.py @@ -0,0 +1,186 @@ +"""Task handoff and answer deadlines, using only the existing temporary test store.""" +import threading +import time +import unittest +from concurrent.futures import ThreadPoolExecutor +from unittest import mock + +from test_ai_tasks import isolate +from api import ai_answer_wait, ai_gates, ai_panels, ai_tasks, mcp_task_tools + + +class HandoffEdges(unittest.TestCase): + def setUp(self): + isolate() + ai_answer_wait._typed.clear() + self.ctx = {"client": "A", "session": "one"} + + def ask(self, recommended=True): + options = [{"id": "a", "label": "A", "recommended": recommended}, {"id": "b", "label": "B"}] + with mock.patch.object(mcp_task_tools, "ASK_WAIT_MIN", 0): + return mcp_task_tools.ask_user(self.ctx, "Which?", options, wait_seconds=0) + + def test_unscoped_timeout_delivers_auto_answer_in_same_call(self): + with mock.patch.object(ai_gates, "answer_wait", return_value=0): + self.ask() + with mock.patch.object(ai_gates, "no_answer", return_value="recommended"): + answer, _ = mcp_task_tools.get_answers(self.ctx) + self.assertEqual(answer["answers"][0]["choice"], ["a"]) + self.assertTrue(answer["answers"][0]["recommended_on_timeout"]) + self.assertEqual(answer["still_waiting"], []) + + def test_unscoped_expired_question_tells_ai_to_stop(self): + with mock.patch.object(ai_gates, "answer_wait", return_value=0): + self.ask(False) + answer, _ = mcp_task_tools.get_answers(self.ctx) + self.assertIs(answer.get("keep_waiting"), False) + self.assertIn("end your turn", answer["note"]) + + def test_user_answer_wins_race_with_auto_pick_and_is_returned_now(self): + with mock.patch.object(ai_gates, "answer_wait", return_value=0): + asked = self.ask() + original = ai_panels.answer + def user_first(task_id, step_id, block_id, **kwargs): + original(task_id, step_id, block_id, choice=["b"]) + return original(task_id, step_id, block_id, **kwargs) + with mock.patch.object(ai_gates, "no_answer", return_value="recommended"), \ + mock.patch.object(ai_panels, "answer", side_effect=user_first): + answer, _ = mcp_task_tools.get_answers(self.ctx, panel=asked["panel"]) + self.assertEqual(answer["answers"][0]["choice"], ["b"]) + self.assertNotIn("recommended_on_timeout", answer["answers"][0]) + self.assertNotIn("end your turn", answer.get("note", "")) + + def test_resume_uses_user_reply_time_not_background_task_update(self): + older = ai_tasks.begin_task("A", "Older", now=100) + ai_panels.leave_message(older["id"], "first", now=110) + newer = ai_tasks.begin_task("A", "Newer", now=120, new_task=True) + ai_panels.leave_message(newer["id"], "second", now=130) + older["updated"] = 140 # a late run monitor / metadata update is not a new user reply + ai_tasks._current.clear() + self.assertEqual(ai_tasks.current_task("A", now=150)["id"], newer["id"]) + + def test_new_session_does_not_take_another_sessions_inflight_task(self): + busy = ai_tasks.begin_task("A", "Busy", now=100, session="one") + ai_panels.leave_message(busy["id"], "keep working", now=101) + running = ai_tasks.begin_step("A", "describe_canvas", {}, 102, True, session="one") + self.assertIsNotNone(running) + other = ai_tasks.current_task("A", now=103, session="two") + self.assertNotEqual(other["id"], busy["id"]) + self.assertEqual(ai_tasks.active_task_id("A", "one"), busy["id"]) + + def test_recent_panel_answer_can_resume_its_task(self): + asked = self.ask() + task, step = ai_panels._find(asked["panel"]) + task["ended"] = time.time() + ai_panels.answer(task["id"], step["id"], "answer", choice=["b"]) + ai_tasks._current.clear() + self.assertEqual(ai_tasks.current_task("A", session="next")["id"], task["id"]) + + def test_typing_defers_automatic_choice(self): + asked = self.ask() + task, step = ai_panels._find(asked["panel"]) + late = step["at"] + ai_gates.answer_wait() + 5 + ai_answer_wait.typing(task["id"], now=late - 1) + with mock.patch.object(ai_gates, "no_answer", return_value="recommended"): + answer = ai_answer_wait.wait_on(asked["panel"], now=late) + self.assertTrue(answer["keep_waiting"]) + self.assertTrue(answer["typing"]) + self.assertFalse(step.get("answers")) + + def test_ineligible_replies_never_resume_a_task(self): + for entry in ({"at": 100}, {"at": 7301}, {"at": 7299, "delivered": True}, + {"at": 7299, "recommended_on_timeout": True}): + with self.subTest(entry=entry): + isolate() + task = ai_tasks.begin_task("A", "Old", now=100) + task["steps"].append({"kind": "panel", "answers": {"answer": entry}}) + ai_tasks._current.clear() + self.assertNotEqual(ai_tasks.current_task("A", now=7300)["id"], task["id"]) + + def test_timeout_with_multiple_panels_delivers_once_and_keeps_other_question(self): + asked = self.ask() + other = self.ask(False) + task, step = ai_panels._find(asked["panel"]) + step["at"] -= ai_gates.answer_wait() + 5 + with mock.patch.object(ai_gates, "no_answer", return_value="recommended"): + answer, _ = mcp_task_tools.get_answers(self.ctx) + again, _ = mcp_task_tools.get_answers(self.ctx) + self.assertEqual(len(answer["answers"]), 1) + self.assertEqual(answer["still_waiting"], [{"panel": other["panel"], "question": "answer"}]) + self.assertEqual(again["answers"], []) + self.assertTrue(again["keep_waiting"]) + + def test_ask_user_returns_a_racing_user_answer(self): + original = ai_panels.answer + def user_first(task_id, step_id, block_id, **kwargs): + original(task_id, step_id, block_id, choice=["b"]) + return original(task_id, step_id, block_id, **kwargs) + with mock.patch.object(ai_gates, "answer_wait", return_value=0), \ + mock.patch.object(ai_gates, "no_answer", return_value="recommended"), \ + mock.patch.object(ai_panels, "answer", side_effect=user_first): + got = self.ask() + self.assertTrue(got["answered"]) + self.assertEqual(got["choice"], ["b"]) + + def test_ask_user_delivers_messages_and_images_received_while_waiting(self): + from api import ai_marks + original = ai_panels.get_answers + sent = [] + def message_first(client, panel_id=None, *args, **kwargs): + if not sent: + task, _ = ai_panels._find(panel_id) + sent.append(ai_panels.leave_message(task["id"], "Stop and read this first")) + return original(client, panel_id, *args, **kwargs) + with mock.patch.object(mcp_task_tools, "ASK_WAIT_MIN", 0), \ + mock.patch.object(ai_panels, "get_answers", side_effect=message_first), \ + mock.patch.object(ai_marks, "image_items", return_value=[{"type": "image"}]) as images: + got, media = mcp_task_tools.TASK_TOOLS["ask_user"]["run"](self.ctx, question="Which?", wait_seconds=0) + self.assertFalse(got["answered"]) + self.assertEqual(got["messages"][0]["id"], sent[0]["id"]) + self.assertEqual(media, [{"type": "image"}]) + images.assert_called_once_with(got["messages"]) + self.assertNotIn("messages", original("A")) + + def test_simultaneous_user_answer_and_timeout_commit_only_one_choice(self): + asked = self.ask() + task, step = ai_panels._find(asked["panel"]) + late = step["at"] + ai_gates.answer_wait() + 5 + start = threading.Barrier(2) + def user(): + start.wait(timeout=5) + try: + ai_panels.answer(task["id"], step["id"], "answer", choice=["b"]) + return True + except ai_panels.PanelError as error: + self.assertEqual(str(error), "Already answered") + return False + def timeout(): + start.wait(timeout=5) + return ai_answer_wait.wait_on(asked["panel"], now=late) + with mock.patch.object(ai_gates, "no_answer", return_value="recommended"), ThreadPoolExecutor(2) as pool: + user_result, timeout_result = pool.submit(user), pool.submit(timeout) + won = user_result.result(timeout=10) + timeout_result.result(timeout=10) + received = ai_panels.get_answers("A", asked["panel"]) + self.assertEqual(len(received["answers"]), 1) + self.assertEqual(received["answers"][0]["choice"], ["b"] if won else ["a"]) + self.assertEqual(bool(received["answers"][0].get("recommended_on_timeout")), not won) + self.assertEqual(ai_panels.get_answers("A", asked["panel"])["answers"], []) + + def test_message_arriving_at_deadline_prevents_an_automatic_choice(self): + asked = self.ask() + task, step = ai_panels._find(asked["panel"]) + step["at"] -= ai_gates.answer_wait() + 5 + original = ai_panels.get_answers + sent = [] + def message_after_snapshot(*args, **kwargs): + snapshot = original(*args, **kwargs) + if not sent: + sent.append(ai_panels.leave_message(task["id"], "Stop, I changed my mind")) + return snapshot + with mock.patch.object(ai_panels, "get_answers", side_effect=message_after_snapshot), \ + mock.patch.object(ai_gates, "no_answer", return_value="recommended"): + result, _ = mcp_task_tools.get_answers(self.ctx) + self.assertEqual(result["messages"][0]["id"], sent[0]["id"]) + self.assertEqual(result["answers"], []) + self.assertFalse(step.get("answers")) diff --git a/tests/test_ai_tasks.py b/tests/test_ai_tasks.py index 4e80b8e..fc5d8ef 100644 --- a/tests/test_ai_tasks.py +++ b/tests/test_ai_tasks.py @@ -15,6 +15,7 @@ sys.path[:0] = [str(PLUGIN), str(PLUGIN.parents[1])] from api import action_catalog, activity_log, ai_callers, ai_panels, ai_task_changes, ai_task_runs, ai_tasks, ai_tasks_routes, mcp_task_tools # noqa: E402 from api.mcp_server import TOOLS # noqa: E402 +from api.ai_task_handoff import REPLY_FRESH TMP = tempfile.mkdtemp(prefix="amb_ai_tasks_") LOG = tempfile.mkdtemp(prefix="amb_ai_activity_") @@ -331,7 +332,7 @@ class TaskTests(unittest.TestCase): ai_tasks.begin_task("A", "前天的") ai_tasks.report("A", "做完了", done=True) old_id = ai_tasks.list_tasks()[0]["id"] - ai_panels.leave_message(old_id, "这根手指怪怪的", now=ai_tasks.time.time() - ai_tasks.REPLY_FRESH - 60) + ai_panels.leave_message(old_id, "这根手指怪怪的", now=ai_tasks.time.time() - REPLY_FRESH - 60) ai_tasks._current.clear() # ComfyUI started again since: no task is this app's open one ai_tasks.record("A", "describe_canvas", {}, ok({}), ai_tasks.time.time(), True) self.assertNotEqual(ai_tasks.get_task(old_id)["steps"][-1]["tool"], "describe_canvas", "a new task, not the one written to days ago")