Fix task resumption and answer delivery at wait deadlines
This commit is contained in:
@@ -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 软件重新取了列表,黄字自己消失。
|
||||
|
||||
+36
-8
@@ -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
|
||||
|
||||
@@ -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
|
||||
+2
-8
@@ -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"):
|
||||
|
||||
+18
-10
@@ -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"},
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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"))
|
||||
@@ -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")
|
||||
|
||||
Reference in New Issue
Block a user