- An AI app calling by an older tool list (tool_help or use_tool not offered, use_tool for a tool offered directly, a tool the list lacks) is told so, and the task header shows how to reconnect until the app fetches tools/list again. - add_lora / set_model compare the base models on the model line (model_advice.py; the page reports the line) and add base_model_note. - get_model gives author_examples (sampler, steps, CFG, size, a LoRA's weight from the Civitai record's example pictures) and the version name. - run_workflow says ComfyUI's refusal as which node, what is wrong and the tool that fixes it (run_refusal.py; mcp_run.js passes the refusal as data). - Gates "When I don't answer": wait for me, or go with the recommendation once the waiting time runs out (choices with a recommended option only), handed to the AI in the same call and shown on the stage and under Decisions. - A tool budget test (count, size, description length, required arguments) and the exam tasks with their scorecard (docs/architecture/tool-exam.md, tools/tool_exam.mjs). The tool set's text recommends Full for Antigravity. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
57 lines
2.9 KiB
Python
57 lines
2.9 KiB
Python
"""ComfyUI's refusal of a run said as what to do next (api/run_refusal.py), in run_workflow's answer."""
|
|
import sys
|
|
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])]
|
|
|
|
from api import mcp_actions, run_refusal # noqa: E402
|
|
from api.mcp_tools import ToolError # noqa: E402
|
|
|
|
|
|
def node(node_id, kind, **extra):
|
|
return {"node": node_id, "class_type": "KSampler", "type": kind, "message": "Problem", "details": "", **extra}
|
|
|
|
|
|
class SayTests(unittest.TestCase):
|
|
def test_each_node_problem_names_the_node_and_the_tool(self):
|
|
text = run_refusal.say({"type": "prompt_outputs_failed_validation", "nodes": [
|
|
node("3", "required_input_missing", input="model"),
|
|
node("4", "value_not_in_list", input="ckpt_name", value="gone.safetensors"),
|
|
node("5", "value_bigger_than_max", input="cfg", value="80", details="max 30"),
|
|
]})
|
|
self.assertIn("node 3 (KSampler) has nothing wired into model: connect_nodes", text)
|
|
self.assertIn("ckpt_name = gone.safetensors is not one of its choices", text)
|
|
self.assertIn("fix_workflow_models", text)
|
|
self.assertIn("cfg = 80 is above its maximum (max 30)", text)
|
|
self.assertTrue(text.endswith("run_workflow again."))
|
|
|
|
def test_the_whole_prompt_and_unknown_kinds(self):
|
|
self.assertIn("SaveImage", run_refusal.say({"type": "prompt_no_outputs", "message": "Prompt has no outputs", "nodes": []}))
|
|
self.assertIn("node 9 (KSampler): Odd thing it said", run_refusal.say({"nodes": [
|
|
{"node": "9", "class_type": "KSampler", "type": "custom_validation_failed", "message": "Odd thing it said"}]}))
|
|
self.assertIn("Server down 500", run_refusal.say({"type": "server_error", "message": "Server down", "details": "500", "nodes": []}))
|
|
self.assertEqual(run_refusal.say(None), "")
|
|
self.assertEqual(run_refusal.say({"nodes": []}), "")
|
|
|
|
def test_many_problems_are_cut_short(self):
|
|
text = run_refusal.say({"nodes": [node(str(n), "required_input_missing", input="model") for n in range(12)]})
|
|
self.assertIn("and 4 more", text)
|
|
|
|
|
|
class RunTests(unittest.TestCase):
|
|
def test_run_workflow_says_what_to_fix(self):
|
|
refused = {"prompt_ids": [], "queued": 0, "requested": 1, "submission_error": "Prompt outputs failed validation",
|
|
"refusal": {"type": "prompt_outputs_failed_validation", "nodes": [node("3", "required_input_missing", input="model")]}}
|
|
with mock.patch.object(mcp_actions.mcp_fill, "fill", return_value=None), \
|
|
mock.patch.object(mcp_actions, "ask_page", return_value=refused):
|
|
with self.assertRaises(ToolError) as caught:
|
|
mcp_actions.run_workflow({})
|
|
self.assertIn("node 3 (KSampler) has nothing wired into model", str(caught.exception))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|