DemonGatanjieuandClaude Opus 5.5
9633d3451a
Gemini 3.8 Flash took seven exam tasks through Antigravity (107 calls, 19
failed, every task done). Nearly all failures were a tool's first use:
empty or guessed arguments. So:
- An argument error lists what the tool takes ("set_prompt takes: prompt
(needed), role, node_id, mode"), and the next call is right.
- One argument name per thing: set_prompt's text is prompt (as in
run_workflow), open_workflow's name is workflow (as in run_workflow).
- A wait over 55 s waits 55 s instead of failing; search_models takes a
model type in the singular too.
- An unread reply older than two hours no longer draws a new chat's calls
into its old task (a two-day-old mark took the first exam task).
The exam doc gets the setup, the Antigravity command line and the results.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
73 lines
3.8 KiB
Python
73 lines
3.8 KiB
Python
"""Attach explicitly identified external ComfyUI submissions to an AI task; never guess ownership."""
|
||
import json
|
||
import math
|
||
import time
|
||
|
||
from . import ai_task_runs, ai_tasks
|
||
from .mcp_tools import ToolError
|
||
|
||
|
||
def checked_wait(value):
|
||
"""wait_seconds as seconds from 0 to 55: one call waits no longer, so a longer wait is cut to that."""
|
||
if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value):
|
||
raise ToolError("wait_seconds is a number of seconds, up to 55.")
|
||
return min(max(value, 0), 55)
|
||
|
||
|
||
def track_run(ctx, prompt_ids, task=None, wait_seconds=15):
|
||
wait_seconds = checked_wait(wait_seconds)
|
||
if (not isinstance(prompt_ids, list) or not 1 <= len(prompt_ids) <= 8
|
||
or any(not isinstance(pid, str) or not pid or len(pid) > 128 for pid in prompt_ids)
|
||
or len(set(prompt_ids)) != len(prompt_ids)):
|
||
raise ToolError("Give 1–8 unique prompt_ids returned by your own ComfyUI /prompt submissions.")
|
||
client, session = ctx.get("client") or "AI", ctx.get("session") or ""
|
||
task_id = task or ctx.get("task") or ai_tasks.active_task_id(client, session)
|
||
with ai_tasks._lock:
|
||
target = ai_tasks._load().get(task_id)
|
||
if not target or target.get("ended"):
|
||
raise ToolError("Begin or resume an open task before attaching a run; pass its task ID.")
|
||
existing = None
|
||
for candidate in ai_tasks._load().values():
|
||
for step in candidate.get("steps", []):
|
||
ids = step.get("prompt_ids") or []
|
||
if set(ids).intersection(prompt_ids):
|
||
if candidate["id"] != task_id or ids != prompt_ids:
|
||
raise ToolError("A prompt ID is already attached to another run. Do not reassign it.")
|
||
existing = step
|
||
if existing is None:
|
||
try:
|
||
queue = ai_task_runs._queue()
|
||
running, pending = queue.get_current_queue()
|
||
known = {item[1] for item in [*running, *pending]}
|
||
missing = [pid for pid in prompt_ids if pid not in known and not queue.get_history(prompt_id=pid).get(pid)]
|
||
except Exception as error:
|
||
raise ToolError("Cannot read ComfyUI's queue/history; nothing was attached.") from error
|
||
if missing:
|
||
raise ToolError("Unknown prompt IDs: " + ", ".join(missing) + ". Use this server's submission receipts.")
|
||
receipt = {"prompt_ids": prompt_ids, "queued": len(prompt_ids)}
|
||
result = {"structuredContent": receipt, "content": []}
|
||
step = ai_tasks.record(client, "run_workflow", {}, result, time.time(), session=session, task_id=task_id)
|
||
if not step:
|
||
raise ToolError("Could not record the run; nothing was queued by this tool.")
|
||
step.update(prompt_ids=list(prompt_ids), source="external")
|
||
ai_tasks._save(target)
|
||
ai_task_runs.watch(step["id"], receipt)
|
||
existing = step
|
||
answer = ai_task_runs.status(client, session, task_id, wait_seconds, step_id=existing["id"])
|
||
return {"task": task_id, "attached": True, **answer}
|
||
|
||
|
||
def wait_for_receipt(result, client, session, task_id, wait_seconds=15, step_id=None):
|
||
"""Return authoritative execution state alongside submission receipts, with a bounded wait."""
|
||
answer = ai_task_runs.status(client, session, task_id, wait_seconds, step_id=step_id)
|
||
data = {**result.get("structuredContent", {}), **answer, "task": task_id}
|
||
result["structuredContent"] = data
|
||
content = list(result.get("content") or [])
|
||
text = {"type": "text", "text": json.dumps(data, ensure_ascii=False)}
|
||
if content and content[0].get("type") == "text":
|
||
content[0] = text
|
||
else:
|
||
content.insert(0, text)
|
||
result["content"] = content
|
||
return result
|