Compare commits

...
1 Commits
Author SHA1 Message Date
Ivor StrelarandClaude Sonnet 5 d3e8e6d470 fix: process library builder in batches to stop the turn/cost blowup
A single session asking Claude to open an entire large library, compose one
manifest, and write it via the Write tool could spiral to 200-400+ turns and
tens of dollars of subscription usage with nothing produced (is_error: true,
reported on a 101-asset library). Rewritten so each Claude call handles only
batch_size assets (default 15) and replies with JSON only — the node itself
assembles and writes the final manifest, so Claude is never given a
file-writing tool for this.

Progress is tracked in a hidden, atomically-written checkpoint
(.sf_library_builder_checkpoint.json, cleared on success). A failed or
interrupted run resumes from the same settings instead of re-paying for
finished batches.

Verified on 45 synthetic assets: 16 turns and ~$0.08 per batch of 15 (vs.
198-401 turns and $25-130 for 101 assets before), and confirmed resume by
pre-seeding a checkpoint and watching it skip the already-done batches while
preserving their entries and reusing the established art_style.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-10-12 10:16:46 +02:00
3 changed files with 314 additions and 149 deletions
+11
View File
@@ -123,6 +123,17 @@ anything. Turn it on once you're happy, or paste the text in yourself.
Start in **manifest only**. Reorganising is genuinely useful on a folder of `IMG_0021.png` and
`final_v2 copy.png`, but it moves real files — try it on a copy first.
**Large libraries are processed in batches, not one session.** `batch_size` (default 15) caps
how many assets Claude handles per call — asking it to hold a whole large library open, compose
one giant manifest, and write the file all in a single uninterrupted session is what used to
blow up in turns and cost on libraries of 100+ assets, with nothing produced at the end. Each
batch now only reads its own assets and replies with JSON; the node assembles and writes the
manifest itself, so no file-writing tool is ever handed to Claude for this.
A hidden checkpoint (`.sf_library_builder_checkpoint.json`, removed automatically on success)
tracks which assets are already catalogued. If a run fails or times out partway, rerunning with
the same settings **resumes** — finished batches are skipped, not re-paid for.
#### Using SF Claude Asset Director
Give it a folder of assets and a brief; it returns the references Claude chose and the prompt
+302 -148
View File
@@ -1,11 +1,11 @@
# ABOUTME: Claude surveys a folder of game assets and writes a searchable manifest for it
# ABOUTME: Can optionally reorganise and rename the files; read-only by default
# ABOUTME: Processes assets in small batches with a resumable checkpoint, so a failure costs little
import hashlib
import json
import os
import re
import shutil
import subprocess
import tempfile
import time
from .sf_claude_code import MODEL_CHOICES, find_claude_binary
@@ -18,27 +18,16 @@ MODE_IN_PLACE = "organise in place"
MODES = [MODE_MANIFEST, MODE_COPY, MODE_IN_PLACE]
SCHEMA_VERSION = "sf-asset-library/1"
CHECKPOINT_NAME = ".sf_library_builder_checkpoint.json"
# What the manifest has to contain for SF Claude Asset Director to search it well:
# one flat list (cheaper to scan than a tree), the same keys on every entry, a
# one-sentence description carrying the meaning, and a project-level art style so
# generated prompts stay consistent across the whole library.
SCHEMA_SPEC = """{
"schema": "sf-asset-library/1",
"project": "<short project name>",
"art_style": "<one sentence describing the shared visual style of the whole library>",
"generated": "<YYYY-MM-DD>",
"assets": [
{
"file": "<path relative to the library root, exactly as it exists on disk>",
"name": "<short human name>",
"category": "<character|environment|prop|ui|vfx|other>",
"subcategory": "<hero|enemy|npc|location|background|weapon|icon|...>",
"archetype": "<for characters: tank|healer|rogue|mage|boss|minion|... else omit>",
"tags": ["<lowercase keyword>", "..."],
"desc": "<one sentence: what it depicts, pose/view, dominant colours>"
}
]
ASSET_SCHEMA_SPEC = """{
"file": "<path relative to the library root, exactly as it exists on disk>",
"name": "<short human name>",
"category": "<character|environment|prop|ui|vfx|other>",
"subcategory": "<hero|enemy|npc|location|background|weapon|icon|...>",
"archetype": "<for characters: tank|healer|rogue|mage|boss|minion|... else omit>",
"tags": ["<lowercase keyword>", "..."],
"desc": "<one sentence: what it depicts, pose/view, dominant colours>"
}"""
DEFAULT_SYSTEM_PROMPT = """You are cataloguing a game asset library.
@@ -51,7 +40,7 @@ library over inventing a new word per asset."""
def scan_assets(root):
"""Image files under `root`, relative to it."""
"""Image files under `root`, relative to it, in a stable sorted order."""
found = []
for folder, dirs, names in os.walk(root):
dirs[:] = [d for d in dirs if not d.startswith(".")]
@@ -61,6 +50,11 @@ def scan_assets(root):
return sorted(found)
def chunk(items, size):
for i in range(0, len(items), size):
yield items[i : i + size]
def extract_json(text):
"""The JSON object out of a reply that may be fenced or prefaced."""
start, end = text.find("{"), text.rfind("}")
@@ -73,38 +67,88 @@ def extract_json(text):
raise RuntimeError(f"Claude's JSON could not be parsed ({e}): {body[:300]}")
def build_instruction(mode, assets, project_notes, manifest_name, write_manifest, out_dir):
"""The task description handed to Claude."""
def canonicalize(claude_path, expected):
"""Map Claude's returned file path onto one of the paths we actually asked
about, tolerating a leading './' or a basename-only answer."""
cleaned = (claude_path or "").strip().lstrip("./")
if cleaned in expected:
return cleaned
base = os.path.basename(cleaned).lower()
for candidate in expected:
if os.path.basename(candidate).lower() == base:
return candidate
return cleaned
def checkpoint_key(mode, manifest_name, system_prompt, project_notes, all_files):
"""A fingerprint of this exact job, so a checkpoint only resumes a matching rerun."""
payload = json.dumps(
{
"mode": mode,
"manifest_name": manifest_name,
"system_prompt": system_prompt,
"project_notes": project_notes,
"files": all_files,
},
sort_keys=True,
)
return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:16]
def load_checkpoint(root, key):
path = os.path.join(root, CHECKPOINT_NAME)
if not os.path.isfile(path):
return None
try:
with open(path, encoding="utf-8") as f:
data = json.load(f)
except (OSError, json.JSONDecodeError):
return None
return data if data.get("key") == key else None
def save_checkpoint(root, key, project, art_style, done_assets):
"""Atomic write, so a crash mid-save can't corrupt the checkpoint."""
path = os.path.join(root, CHECKPOINT_NAME)
data = {"key": key, "project": project, "art_style": art_style, "done_assets": done_assets}
fd, tmp = tempfile.mkstemp(dir=root, prefix=".sf_checkpoint_")
try:
with os.fdopen(fd, "w", encoding="utf-8") as f:
json.dump(data, f)
os.replace(tmp, path)
except Exception:
if os.path.exists(tmp):
os.remove(tmp)
raise
def clear_checkpoint(root):
path = os.path.join(root, CHECKPOINT_NAME)
if os.path.isfile(path):
os.remove(path)
def build_batch_instruction(mode, batch_files, total_count, project_notes, established, out_dir):
"""The task description for ONE batch call — scoped to just its own files."""
parts = [
f"You are building an asset library manifest for {len(assets)} image file(s)."
f"You are cataloguing {len(batch_files)} file(s) from one batch of a "
f"{total_count}-file game asset library. You only need to handle the "
f"files listed below — nothing else exists from your point of view."
]
if project_notes.strip():
parts.append(f"About this project:\n{project_notes.strip()}")
listing = "\n".join(f" {name}" for name in assets)
parts.append(f"The files, relative to your working directory:\n{listing}")
listing = "\n".join(f" {name}" for name in batch_files)
parts.append(f"Your files for this batch:\n{listing}")
parts.append(
"Open each image with the Read tool before describing it. Work through them "
"Open each one with the Read tool before describing it. Work through them "
"steadily; do not guess from a filename."
)
if mode == MODE_MANIFEST:
parts.append(
"Do NOT move, rename, copy or delete anything. This is a survey only."
)
if write_manifest:
parts.append(
f"When you have seen them all, write the manifest to `{manifest_name}` "
f"in the working directory using the Write tool, then reply with the "
f"word DONE and nothing else."
)
else:
parts.append(
"When you have seen them all, reply with ONLY the manifest JSON — no "
"commentary, no code fence."
)
parts.append("Do NOT move, rename, copy or delete anything. This is a survey only.")
else:
where = (
f"a new folder at `{out_dir}`, leaving the originals untouched"
@@ -112,22 +156,43 @@ def build_instruction(mode, assets, project_notes, manifest_name, write_manifest
else "the working directory itself"
)
parts.append(
f"Then reorganise the library into {where}:\n"
f"- Group files into folders by category, e.g. characters/heroes, "
f"characters/enemies, environments, props, ui, vfx.\n"
f"- Rename to lowercase_with_underscores, descriptive of the subject, no "
f"spaces and no version noise. Keep the original extension.\n"
f"- Never discard a file. Every input must end up somewhere.\n"
f"- Use Bash (cp for a copy, mv in place) to do the moving."
f"Then reorganise ONLY these files into {where}:\n"
f"- Group by category, e.g. characters/heroes, characters/enemies, "
f"environments, props, ui, vfx.\n"
f"- Rename to lowercase_with_underscores, descriptive of the subject, "
f"no spaces, no version noise. Keep the original extension.\n"
f"- Never discard a file.\n"
f"- Use Bash (cp for a copy, mv in place) to do the moving.\n"
f"The `file` field of each asset must be its path AFTER reorganising, "
f"relative to the library root."
)
if established:
project, art_style = established
parts.append(
f"The `file` field of every asset must be the path AFTER reorganising, "
f"relative to the library root. Write the manifest to `{manifest_name}` at "
f"the root of the reorganised library, then reply with the word DONE and "
f"nothing else."
f"This library's project name and art style are already settled by an "
f"earlier batch — reuse them exactly, do not invent new ones:\n"
f' project: "{project}"\n art_style: "{art_style}"'
)
parts.append(
f"Reply with ONLY a JSON object of this shape — no commentary, no code "
f"fence, no other top-level keys:\n"
f'{{"assets": [{ASSET_SCHEMA_SPEC}]}}\n'
f"The assets array must contain exactly one entry per file listed above."
)
else:
parts.append(
f"You are establishing this library's project name and art style for "
f"the first time — later batches will reuse whatever you choose here, "
f"so make `art_style` a precise sentence covering the whole library's "
f"visual approach, not just these files.\n"
f"Reply with ONLY a JSON object of this shape — no commentary, no code "
f"fence:\n"
f'{{"project": "<short project name>", "art_style": "<one sentence>", '
f'"assets": [{ASSET_SCHEMA_SPEC}]}}\n'
f"The assets array must contain exactly one entry per file listed above."
)
parts.append(f"The manifest must match this shape exactly:\n{SCHEMA_SPEC}")
return "\n\n".join(parts)
@@ -136,6 +201,11 @@ class SFClaudeLibraryBuilder:
Surveys a folder of game assets and writes a manifest describing each one,
so SF Claude Asset Director can choose between them without opening images.
Processes assets in small batches rather than one session, because a single
session covering a large library can blow up in turns and cost with nothing
to show for it. A checkpoint records finished batches, so a failed or
interrupted run resumes instead of re-paying for work already done.
Read-only by default. It can also reorganise and rename the files, into a
copy or in place, when you ask it to.
"""
@@ -146,11 +216,7 @@ class SFClaudeLibraryBuilder:
"required": {
"library_path": (
"STRING",
{
"multiline": False,
"default": "",
"tooltip": "Folder holding the assets to catalogue.",
},
{"multiline": False, "default": "", "tooltip": "Folder holding the assets to catalogue."},
),
"mode": (
MODES,
@@ -166,11 +232,7 @@ class SFClaudeLibraryBuilder:
),
"output_path": (
"STRING",
{
"multiline": False,
"default": "",
"tooltip": "Destination folder, for 'organise into a copy' only.",
},
{"multiline": False, "default": "", "tooltip": "Destination folder, for 'organise into a copy' only."},
),
"project_notes": (
"STRING",
@@ -193,10 +255,7 @@ class SFClaudeLibraryBuilder:
),
"manifest_name": (
"STRING",
{
"default": "manifest.json",
"tooltip": "Filename for the manifest, written at the library root.",
},
{"default": "manifest.json", "tooltip": "Filename for the manifest, written at the library root."},
),
"write_manifest": (
"BOOLEAN",
@@ -208,6 +267,20 @@ class SFClaudeLibraryBuilder:
),
},
),
"batch_size": (
"INT",
{
"default": 15,
"min": 3,
"max": 50,
"step": 1,
"tooltip": (
"Assets per Claude call. Keeps each session small enough to "
"finish reliably — a failed batch only costs that batch, not "
"the whole library, and a rerun resumes rather than restarts."
),
},
),
"model": (
MODEL_CHOICES,
{"default": "default", "tooltip": "Which Claude model does the cataloguing."},
@@ -215,14 +288,11 @@ class SFClaudeLibraryBuilder:
"timeout_seconds": (
"INT",
{
"default": 1800,
"default": 300,
"min": 60,
"max": 7200,
"step": 60,
"tooltip": (
"Allow roughly 5 seconds per asset. 100 assets needs about "
"10 minutes."
),
"max": 1800,
"step": 30,
"tooltip": "Per BATCH, not per library. Budget roughly 5s per asset in the batch.",
},
),
"seed": (
@@ -249,6 +319,11 @@ category, archetype, tags and a one-line description — plus a project-level ar
style. **SF Claude Asset Director** then searches that manifest instead of
opening images.
Processes **batch_size** assets per Claude call rather than the whole library at
once, and tracks progress in a hidden checkpoint file. If a run fails partway,
rerunning with the same settings resumes from where it stopped instead of
re-paying for batches already done.
Defaults to **manifest only**, which touches nothing: you get the JSON as text to
read and save yourself. Switch `write_manifest` on to have it saved into the
library, or change `mode` to let Claude reorganise and rename the files too.
@@ -256,8 +331,20 @@ library, or change `mode` to let Claude reorganise and rename the files too.
Budget about 5 seconds per asset — it opens every image.
"""
def build(self, library_path, mode, output_path, project_notes, system_prompt,
manifest_name, write_manifest, model, timeout_seconds, seed):
def build(
self,
library_path,
mode,
output_path,
project_notes,
system_prompt,
manifest_name,
write_manifest,
batch_size,
model,
timeout_seconds,
seed,
):
root = os.path.expanduser(library_path.strip())
if not root or not os.path.isdir(root):
raise ValueError(f"library_path is not a folder: {library_path!r}")
@@ -281,94 +368,161 @@ Budget about 5 seconds per asset — it opens every image.
if not assets:
raise ValueError(f"No image files found in {root}")
writes_file = write_manifest or mode != MODE_MANIFEST
tools = "Read" if mode == MODE_MANIFEST and not write_manifest else "Read,Write,Bash"
key = checkpoint_key(mode, manifest_name, system_prompt, project_notes, assets)
checkpoint = load_checkpoint(root, key)
done_by_file = {}
project = art_style = None
if checkpoint:
project = checkpoint.get("project")
art_style = checkpoint.get("art_style")
for entry in checkpoint.get("done_assets", []):
if entry.get("file"):
done_by_file[entry["file"]] = entry
print(
f"[SF Claude Library Builder] Resuming — {len(done_by_file)}/{len(assets)} "
f"asset(s) already catalogued from a previous run."
)
command = [
binary,
"-p",
build_instruction(mode, assets, project_notes, manifest_name,
writes_file, out_dir),
"--output-format",
"json",
"--allowedTools",
tools,
]
if system_prompt.strip():
command += ["--append-system-prompt", system_prompt.strip()]
if model != "default":
command += ["--model", model]
# Claude only ever reads images and replies with JSON; the node writes the
# manifest itself, and the Write tool is never offered. Bash is needed
# only to move/rename files in the organise modes.
tools = "Read" if mode == MODE_MANIFEST else "Read,Bash"
batches = list(chunk(assets, max(1, batch_size)))
print(
f"[SF Claude Library Builder] Cataloguing {len(assets)} asset(s) — mode='{mode}', "
f"tools={tools}. This opens every image, so allow time."
)
total_cost = 0.0
started = time.time()
try:
completed = subprocess.run(
command,
cwd=root,
capture_output=True,
text=True,
timeout=timeout_seconds,
stdin=subprocess.DEVNULL,
)
except subprocess.TimeoutExpired:
raise RuntimeError(
f"Claude did not finish within {timeout_seconds}s. Raise timeout_seconds "
f"(about 5s per asset) or catalogue a smaller folder."
for batch_num, batch_files in enumerate(batches, start=1):
remaining = [f for f in batch_files if f not in done_by_file]
if not remaining:
continue # fully covered by the checkpoint already
established = (project, art_style) if project and art_style else None
instruction = build_batch_instruction(
mode, remaining, len(assets), project_notes, established, out_dir
)
if completed.returncode != 0:
detail = (completed.stderr or completed.stdout or "").strip()[:500]
raise RuntimeError(f"Claude Code exited with code {completed.returncode}. {detail}")
command = [binary, "-p", instruction, "--output-format", "json", "--allowedTools", tools]
if system_prompt.strip():
command += ["--append-system-prompt", system_prompt.strip()]
if model != "default":
command += ["--model", model]
try:
payload = json.loads(completed.stdout)
except json.JSONDecodeError:
raise RuntimeError(f"Unreadable response: {completed.stdout.strip()[:400]}")
print(
f"[SF Claude Library Builder] Batch {batch_num}/{len(batches)} "
f"({len(remaining)} asset(s))..."
)
if payload.get("is_error"):
raise RuntimeError(f"Claude reported an error: {payload.get('result', 'unknown')}")
def _fail(message):
save_checkpoint(root, key, project, art_style, list(done_by_file.values()))
raise RuntimeError(
f"{message} {len(done_by_file)}/{len(assets)} asset(s) are saved in a "
f"checkpoint — rerun with the same settings to resume rather than "
f"start over."
)
try:
completed = subprocess.run(
command,
cwd=root,
capture_output=True,
text=True,
timeout=timeout_seconds,
stdin=subprocess.DEVNULL,
)
except subprocess.TimeoutExpired:
_fail(
f"Batch {batch_num}/{len(batches)} did not finish within "
f"{timeout_seconds}s."
)
if completed.returncode != 0:
detail = (completed.stderr or completed.stdout or "").strip()[:500]
_fail(f"Batch {batch_num}/{len(batches)} failed (exit {completed.returncode}). {detail}")
try:
payload = json.loads(completed.stdout)
except json.JSONDecodeError:
_fail(
f"Batch {batch_num}/{len(batches)}: unreadable response: "
f"{completed.stdout.strip()[:300]}"
)
if payload.get("is_error"):
_fail(
f"Batch {batch_num}/{len(batches)} reported an error: "
f"{payload.get('result', 'unknown')}."
)
cost = payload.get("total_cost_usd") or 0.0
total_cost += cost
turns = payload.get("num_turns", "?")
data = extract_json(payload.get("result", ""))
if project is None:
project = data.get("project", "Untitled project")
art_style = data.get("art_style", "")
expected = set(remaining)
for entry in data.get("assets") or []:
raw = entry.get("file")
if not raw:
continue
entry["file"] = canonicalize(raw, expected)
done_by_file[entry["file"]] = entry
save_checkpoint(root, key, project, art_style, list(done_by_file.values()))
print(
f"[SF Claude Library Builder] Batch {batch_num}/{len(batches)} done: "
f"{turns} turns, ${cost:.2f}. Running total: "
f"{len(done_by_file)}/{len(assets)} asset(s), ${total_cost:.2f}."
)
elapsed = time.time() - started
reply = payload.get("result", "")
# When Claude wrote the file, read it back; otherwise its reply IS the manifest.
# Assemble the final manifest in the library's original scan order.
final_assets = [done_by_file[f] for f in assets if f in done_by_file]
manifest_data = {
"schema": SCHEMA_VERSION,
"project": project or "Untitled project",
"art_style": art_style or "",
"generated": time.strftime("%Y-%m-%d"),
"assets": final_assets,
}
manifest_text = json.dumps(manifest_data, indent=2)
manifest_root = out_dir if mode == MODE_COPY else root
manifest_path = os.path.join(manifest_root, manifest_name)
if write_manifest:
os.makedirs(manifest_root, exist_ok=True)
with open(manifest_path, "w", encoding="utf-8") as f:
f.write(manifest_text)
if writes_file:
if not os.path.isfile(manifest_path):
raise RuntimeError(
f"Claude was asked to write {manifest_name} but it is not there. "
f"It replied: {reply.strip()[:300]}"
)
with open(manifest_path, encoding="utf-8") as f:
manifest_text = f.read()
else:
manifest_text = json.dumps(extract_json(reply), indent=2)
clear_checkpoint(root)
data = extract_json(manifest_text)
entries = data.get("assets") or []
counts = {}
for entry in entries:
for entry in final_assets:
counts[entry.get("category", "?")] = counts.get(entry.get("category", "?"), 0) + 1
report = "\n".join([
f"mode: {mode}",
f"scanned: {len(assets)} file(s) in {root}",
f"catalogued: {len(entries)} asset(s)",
f"categories: " + (", ".join(f"{k} x{v}" for k, v in sorted(counts.items())) or "none"),
f"art style: {data.get('art_style', '(not set)')}",
f"manifest: " + (manifest_path if writes_file else "returned as text, not saved"),
f"took: {elapsed:.0f}s",
])
print(f"[SF Claude Library Builder] {len(entries)} asset(s) catalogued in {elapsed:.0f}s.")
report = "\n".join(
[
f"mode: {mode}",
f"scanned: {len(assets)} file(s) in {root}",
f"catalogued: {len(final_assets)} asset(s) across {len(batches)} batch(es)",
"categories: " + (", ".join(f"{k} x{v}" for k, v in sorted(counts.items())) or "none"),
f"art style: {manifest_data['art_style'] or '(not set)'}",
"manifest: " + (manifest_path if write_manifest else "returned as text, not saved"),
f"took: {elapsed:.0f}s, ${total_cost:.2f} total",
]
)
print(
f"[SF Claude Library Builder] Done: {len(final_assets)}/{len(assets)} asset(s), "
f"{elapsed:.0f}s, ${total_cost:.2f}."
)
if len(entries) < len(assets):
report += f"\nWARNING: {len(assets) - len(entries)} file(s) were not catalogued."
if len(final_assets) < len(assets):
report += f"\nWARNING: {len(assets) - len(final_assets)} file(s) were not catalogued."
return (manifest_text, report)
+1 -1
View File
@@ -1,7 +1,7 @@
[project]
name = "comfyui-sf-nodes"
description = "Custom nodes for ComfyUI including LLM Chat with support for Claude, Gemini, and OpenAI models, WaveSpeed AI image/video generation, VertexAI integration, and workflow utilities"
version = "1.3.0"
version = "1.3.1"
license = { file = "LICENSE" }
dependencies = [
"requests>=2.31.0",