Compare commits

..
Author SHA1 Message Date
Vito Sansevero 7b59667a1d perf(maintenance): evaluate each unused-prompt protection once (#586) 2026-10-31 22:46:01 -07:00
Vito Sansevero 680afb5e20 fix(maintenance): list each protected item with its reason in Activity (#586)
Duplicate prompts, unused components, missing links, workflow storage and
rotated logs now record one protected row per kept item, in the review's
wording, and the job report lists every one of them. Kept links and log
files keep their path. The duplicate file cleanup records each copy the
review protected in its Kept group.
2026-10-31 22:24:59 -07:00
Vito 3295ff8b20 Pair category repair (#550) and saved instructions + dialog picker (#431) (#590)
* feat(maintenance): repair old Pair prompts' empty category and tags (#550)

Squashed from PR #579. Closes #550.

* feat(tagging): saved instructions with a default per engine (#431)

Squashed from PR #587.

* feat(tagging): pick saved instructions in the tagging dialogs (#431)

Squashed from PR #588. Closes #431.
2026-11-01 03:01:07 +00:00
Vito 9640ca86f8 feat(tagging): skipped tags hide matching suggestions (#431) (#585)
* feat(tagging): skipped tags hide matching suggestions (#431)

A Skipped tags list in System > Auto-tagging (D202) hides every suggested
tag whose words contain an entry's words, on every suggestion path
(Tag image, Auto-tag selected, the System batch, the legacy single and
batch routes), before the limit. Jobs snapshot the list at their start.
A read-only preview route counts what a list hides among the library's
tags and the downloaded WD14 label lists. The draft store compares list
and object drafts by value (sameDraft).

* test(tagging): count the Skipped tags block as one concept difference (#431)

* fix(tagging): skip a damaged WD14 label list in the preview (#431)

* test(tagging): classify the Skipped tags preview route as reading no media (#431)
2026-11-01 02:06:42 +00:00
Vito 8a81114dfc perf(images): run image work on two own pools and trim after jobs (#523) (#582)
* perf(images): read PNG text without decoding pixels (#523)

Pillow's PngImageFile.text loads every pixel to reach text chunks written
after IDAT. The scan's metadata batch and the image monitor now read PNG
text through utils/png_text.py, which runs Pillow's own chunk handlers
after the first IDAT and seeks over the pixel data.

The legacy .text read is the floor: a differential test compares both
readers on every fixture image, a curated corpus and a seeded fuzz. The
only difference is a file whose pixel data is undecodable or cut short:
it now keeps the text read before its pixels instead of reading nothing.
Animated PNGs still use .text.

* test(images): cover the text budget, XMP and layout guard of the PNG text walk (#523)

* perf(images): run image work on two own pools and trim after jobs (#523)

Image work no longer spreads over the default executor's threads, where
glibc kept an image-sized malloc arena per thread.

- Interactive image work (view previews behind their gate, workspace card
  facts, both inspector routes) runs on the API's fixed pm-image pool of
  3 threads; the on-demand preview pool is merged into it.
- map_parallel (scan metadata, thumbnail and convert jobs, the legacy
  gallery generation, capture previews) shares one long-lived pm-worker
  pool of Worker threads, reused between jobs. The package teardown stops
  it within its bound (part "workers").
- After a scan, thumbnail or convert job, glibc's malloc_trim(0) returns
  free pages once; a no-op on Windows, macOS and musl.
- tools/perf/memory_drive.py measures RSS, heap, threads and arenas under
  the #523 load on the disposable browser test server (manual only).

* fix(images): start no image thread after the teardown; join retired job pools (#523)
2026-11-01 01:24:27 +00:00
Vito 7a2a04b752 feat(send): send a prompt to all eligible nodes of a window (#542) (#578)
* feat(send): send a prompt to all eligible nodes of a window (#542)

The Destination node select offers "All N eligible nodes" when a window has
two or more nodes that can receive the prompt. It is never selected for you.
Apply previews every node first and writes nothing if one refuses; then it
sends one node after another in node id order, each later node with a fresh
ticket checked against the reviewed text, and reports "Sent to X of N"
naming each refusal. A lost reply stops the run and names the node whose
outcome is unknown. Recorded as D191.

* fix(send): refresh held tickets and compare the reviewed composition for All (#542)

A review held longer than the host's 30 s ticket lifetime would have refused
the first node while sending the rest. Every node now takes a fresh ticket
except the first of a pass taken in the same click, and that ticket must
show the reviewed Before texts and composition. The revision is checked
until something has been posted. Node review sections are labelled by
their h4 heading.
2026-11-01 00:41:13 +00:00
Vito cb94b12c1b fix(system): keep a restore review's upload form alive until it is answered (#583) (#584)
* fix(system): keep a restore review's upload form alive until it is answered (#583)

WebKit can read a FormData's file part after fetch() returned, and with a
request interceptor armed it does. uploadRestore held its FormData only in a
local that was dead after the call, so a garbage collection in between dropped
the form and the server got an empty file ("Uploaded file is empty"); the
Restore preview never opened. The page now holds each upload's form until the
upload is answered.

* docs(system): say what the #583 regression spec proves in each browser (#583)
2026-11-01 00:41:12 +00:00
Vito 3cb90b0c1e perf(images): read PNG text without decoding pixels (#523) (#577)
* perf(images): read PNG text without decoding pixels (#523)

Pillow's PngImageFile.text loads every pixel to reach text chunks written
after IDAT. The scan's metadata batch and the image monitor now read PNG
text through utils/png_text.py, which runs Pillow's own chunk handlers
after the first IDAT and seeks over the pixel data.

The legacy .text read is the floor: a differential test compares both
readers on every fixture image, a curated corpus and a seeded fuzz. The
only difference is a file whose pixel data is undecodable or cut short:
it now keeps the text read before its pixels instead of reading nothing.
Animated PNGs still use .text.

* test(images): cover the text budget, XMP and layout guard of the PNG text walk (#523)
2026-11-01 00:04:27 +00:00
Vito 9a43b67008 test: reset a disposed fixture package's exit actions (#580) (#581)
load_package imports the package under a test-only name; its exit
actions stayed registered after cleanup_package removed it from
sys.modules, so the suite ended with a ModuleNotFoundError from
_kill_encoders_at_exit. cleanup_package now resets the package's
exit signal.
2026-10-31 23:17:17 +00:00
62 changed files with 4658 additions and 260 deletions
+129 -9
View File
@@ -349,9 +349,12 @@ and the time left. The steps, in order:
observer join get what is left, which can be 0 s. `stop()` joins watchdog's
emitters without a limit, so it runs on a daemon thread and is waited for
only until the deadline; a later stop waits for that same call (#561).
3. `close_auto_thumbnails(final=True)` stops the preview worker (30 s) and
kills ffmpeg.
4. `PromptTracker.shutdown` stops the prompt cleanup thread (2 s).
3. `close_auto_thumbnails(final=True)` stops the preview worker (30 s),
kills ffmpeg and shuts the interactive image pool down.
4. `shutdown_job_pool` (`utils/parallel.py`) cancels the long-lived job pool's
queued items and joins its threads (5 s; part `workers`, only when a job
used the pool, #523).
5. `PromptTracker.shutdown` stops the prompt cleanup thread (2 s).
A failing step, or a second Ctrl+C, is logged with its traceback and does not
skip the later steps. A clean run logs one INFO line to PromptManager's log:
@@ -456,6 +459,38 @@ A hot SQLite read that returns many rows aggregates them into one step
(`json_group_array`) instead of stepping once per row, so it does not wait for
the GIL once per row while ComfyUI generates (#354, `get_image_link_matches`).
A metadata read never decodes pixels (#523). Pillow's `PngImageFile.text` loads
every pixel to reach text chunks after IDAT; the scan (`_extract_comfyui_metadata`)
and the image monitor (`ComfyUIMetadataExtractor.extract_metadata`) read PNG text
through `utils/png_text.py`, which runs Pillow's own chunk handlers after the
first IDAT and seeks over pixel data. The legacy `.text` read is the floor:
`tests/test_png_text.py` compares both readers on every image under
`tests/fixtures`, a curated corpus (text after IDAT, zTXt, iTXt, split IDAT, bad
checksums, truncations, oversized text, APNG) and a seeded fuzz, and asserts that
neither reader decodes pixels by patching `ImageFile.load`. The only reads that
differ are files whose pixel data is undecodable or cut short: legacy found
nothing, the walk keeps the text before the pixels. Animated PNGs still use
`.text`.
Image work runs on PromptManager's own threads (#523). glibc keeps each thread's
malloc arena at its high-water mark, so image buffers spread over the default
executor's threads stay resident. Interactive image work (view previews behind their
gate, workspace card facts, both inspector routes) runs on the API's fixed
`pm-image` pool (`IMAGE_POOL_SIZE` 3, `_run_image`, `_map_image`; card facts already
cached answer without a trip; while the gate's 3 preview writes hold every thread, a
card or inspector read waits for one of them, by design of the merged pool; after the
package teardown the pool is not recreated and late calls use the default executor). Jobs (scan metadata and file facts, thumbnail and
convert jobs, the legacy gallery generation, capture previews) run through
`map_parallel` on one long-lived `pm-worker` pool of Worker threads, reused between
jobs; a new Worker threads value replaces it (the teardown also joins a replaced pool
still finishing), and a call made from a job thread runs inline. After a scan, thumbnail or convert job, or the legacy gallery generation,
`release_free_memory` (`utils/native_memory.py`) calls glibc's `malloc_trim(0)` once;
it does nothing on Windows, macOS and musl and never raises. No allocator setting
changes. `tests/test_image_pools.py` asserts the thread of each read and write and the
trim; `tests/test_utils_parallel.py` the pool's reuse, resize, nesting and shutdown.
`tools/perf/memory_drive.py` (manual, Linux, never in CI) measures RSS, heap, threads
and arenas under the #523 load against the disposable browser test server.
The `perf budget` job in `.github/workflows/test.yml` is the reference
measurement: `ubuntu-latest`, one Python, no coverage, only the wall-clock tests.
It runs on every push to `main` and on `workflow_dispatch`, never on pull
@@ -1851,8 +1886,8 @@ Inspect plan keeps D60–D66 (plan D60 → D83, D61 → D84, D65 → D88, D68
engines in the mock's order with legacy labels and sizes (`WD14 SwinV2 · ~400
MB`, `WD14 ViT · ~400 MB`, `GGUF JoyCaption (Recommended) · ~4.5 GB` in the
mock's Ollama slot, `HuggingFace JoyCaption · ~16 GB`); the instructions start
from the server's `default_prompt` (since #431 the saved engine's default
instructions, GGUF's for a WD14 engine).
from the chosen engine's default saved instructions (#431, D204; before, the
server's `default_prompt`).
- **D91** (plan D68, ruling T3-bulk): a **Suggestions** number field (1–100) for
every model, prefilled from `autotag.defaults.suggestion_limit` (10); a model
change keeps it.
@@ -2944,7 +2979,8 @@ change: ComfyUI needs a restart.
A Pair save, by a run or the node's Save pair button, writes the category and tags to the positive prompt only when that
save creates the prompt row (maintainer ruling). An existing prompt never changes on a run: empty, cleared or set by a person, including one first seen as
another pair's negative. The negative stays empty, the saved pair keeps its own metadata, and `save_prompt_pair` writes
no prompt metadata unless called with `fill_new_positive`. No backfill; optional repair is #550.
no prompt metadata unless called with `fill_new_positive`. No backfill on a run; the opt-in repair for prompts older
Pair runs saved is System › Maintenance › "Pair category and tags" (#550, "Pair category and tags repair" below).
Tests: `tests/test_prompt_pair_node.py`, `tests/test_api_pairs.py`, `tests/comfyui/pair-node.spec.mjs`. A Python change: ComfyUI needs a restart.
### Delete… on the archived shelf (#492)
@@ -3859,8 +3895,8 @@ workspace never asks; previews are a cache the server keeps:
- `GET /images/serve/thumbnails/<rel>_thumb<ext>` for a missing preview is refused as
text first (traversal, a backslash, a drive or stream colon, empty or dot segments, so
`D:/x`, `\\server\share` and a leading `/` never reach a joined path), then admitted
on the event loop (`_PreviewGate`): 3 previews are written at once on the API's own
`pm-preview` pool (never the default executor), one flight runs per file (the history
on the event loop (`_PreviewGate`): 3 previews are written at once on the API's
interactive `pm-image` pool (never the default executor; #523), one flight runs per file (the history
route shares the flight key and the admission), and at most 64 requests wait; one
more gets its fallback at once without listing a folder. Inside the admission the
preview maps back to its media file under the same root (exact names, then one folder
@@ -3893,7 +3929,7 @@ workspace never asks; previews are a cache the server keeps:
`exclusive` job runs (`_auto_thumbnail_blocker`), checked when a write starts (after
its slot) and again just before its rename.
- Shutdown is bounded and logged: the worker stops at a file boundary, ffmpeg is
killed, an idle worker leaves (the next output starts a fresh one), the view pool
killed, an idle worker leaves (the next output starts a fresh one), the image pool
shuts down, and after the app's cleanup a late monitor event
starts nothing. A package reload stops the workers the previous import left.
- Cards and strips load previews only: Images in every layout and size, Library cards,
@@ -4273,6 +4309,12 @@ the per-section hand-off panel, was retired by S12 (#329): every section is nati
primary stays the concept's plain button; the plan it opens has the primary.
`duplicates dialog matches the concept` and `duplicates comparison detects mutations`
(system-maintenance-concept.spec) pin the mapping.
- **D192** (Maintenance, #550, beyond the concept): Database adds "Pair category and
tags" after "Unused prompt components", a plain Review plan row (no danger mark: it
deletes nothing) with the line "Fill an empty category or tags on prompts older Pair
node runs saved, from their saved pairs. Only empty fields of never-edited prompts."
`system-maintenance-concept.spec.mjs` maps it (`mapPairRow`) as a copy of Optimize
database's row.
- **D102** (Maintenance, S7, Q5, #222): Metadata cache clears only the browser entries
Prompt Manager owns as a cache (`web/js/cache-ownership.js`). The inventory of every
storage writer has none (see "Browser cache ownership" below), so the row's line adds
@@ -4370,6 +4412,23 @@ the per-section hand-off panel, was retired by S12 (#329): every section is nati
choice. The host's checks still refuse a linked field or a changed workflow, and the
sender also refuses if the destination's `revision` (node object plus graph version)
moved since discovery; it is undefined on an older tab, which skips the comparison. A host error reads as a refusal, not "may have applied".
- **D191** (#542, beyond the concept; maintainer 2026-10-31, Q542-1 A to 5 A): a window with two or more
eligible nodes adds a last Destination node option "All N eligible nodes" (`ALL_NODES` in
`prompt-sender.mjs`), never selected for you (D173). Eligible is D173's filter; the nodes it leaves out
stay counted in the discovery sentence. One window only. Preview stays optional (D175): Preview changes
shows one `section.prompt-sender-node` per node, headed "#<id> <title>", in send order (node id,
numeric). Apply previews every node first and writes nothing if any refuses ("#<id> <title>: <reason>
Nothing was sent."). Then it sends one node after another. Only the first node of a pass taken in that
click keeps its ticket; every other node, and every node of a held Preview (host tickets expire after
30 s), takes a fresh ticket that is refused unless its Before texts and composition equal the reviewed
ones. That fresh preview checks the discovered revision until something has been posted, and not after,
since our own write may move the graph version. Refusals are best effort: "Sent to 2 of 3 nodes in
<window>. #12 <title>: <host's refusal>."; Apply stays available and a retry sends to every node again
(rewriting the same text). A lost reply stops the run, names that node as unknown and lists the nodes
not sent. Closing the dialog stops after the current node. The review sections are labelled by their
`h4` node heading. Unit tests in `tests/js/prompt-sender.test.js`, a fake three-node host in
`tests/browser/prompt-inspector.spec.mjs`, real Classic and Nodes 2.0 widgets in
`tests/comfyui/prompt-sender.spec.mjs`.
- **D106** (Integrations, S10, Q8): the send bridge always asks for the ComfyUI window
and node, and the nodes always follow ComfyUI's own renderer, so "Default send
destination" offers only "Ask each time" and "Host compatibility" only "Follow host
@@ -4619,6 +4678,26 @@ the per-section hand-off panel, was retired by S12 (#329): every section is nati
tokens, the two rows on the concept's select row) and `saved instructions: new, edit,
per-engine default, delete, save and reload (D203)`, `a refused instructions save names
the entry and highlights the block (D203)` (workspace-system-tagging.spec).
- **D204** (Tag image and Auto-tag selected dialogs, #431, beyond the concept): for a
language-model engine a "Saved instructions" select (`#image-tag-instructions-pick`,
`#bulk-tag-instructions-pick`) sits above "Tagging instructions" (the mock's state 3),
on the concept's select field: Built-in, then the saved entries (`GET /autotag/models`
`instructions`, full text), the engine's default marked "<name> · default". It starts
at that default; a pick replaces the text. The text stays editable for the run: while
it differs from the picked entry the select shows a disabled "Edited for this run" and
a muted line under the textarea reads "Edits apply to this run only. Manage saved
instructions in System › Auto-tagging." Choosing an entry again replaces the edit.
Switching between GGUF and HF starts from the new engine's default and drops an edit;
WD14 hides both. A preset (Preview tagging review, the bulk review) shows the entry
with its text, else "Edited for this run". When the read fails the select holds only
Built-in with the client's copy (`BUILTIN_INSTRUCTIONS`, pinned equal to
`DEFAULT_TAG_TEMPLATE` by workspace-image-tagging-state.test). Request bodies are
unchanged (`options.instructions` is the textarea's text); there is no "Save as new".
The workspace no longer reads `default_prompt`. Pinned by the tagging comparison's
mapping (tagging-canonical-comparison.mjs: the select before the instructions),
`the saved instructions picker fills, edits one-off and resets per engine (D204)`
(workspace-tagging.spec) and `Auto-tag selected picks saved instructions and sends the
text shown (D204)` (workspace-tagging-bulk.spec).
- **D205** (Auto-tagging, #431; amends D134): for a language-model engine the batch
review adds a read-only "Instructions" row with the drafted engine default's name
before "Settings", which reads "Up to L tags per image"; the batch and Preview tagging
@@ -4855,6 +4934,47 @@ replaced the database since the review; the plan is dropped). The job:
prompt ids only), logs the backup by its file name, and prunes safety backups to the
newest 10. With no backup the log and the status line say "No safety backup".
#### Pair category and tags repair (#550)
`pair_metadata` ("Pair category and tags", D192; `database/pair_metadata_repair.py`) fills
the empty category and the empty tag set of prompts that Pair node runs before #546
saved, from their saved pairs. It is reviewed and applied like the operations above and
follows the safety backup policy like them (no forced backup). Rules
(`docs/egm/post-3.4.1/550-design.md`):
- Considered: prompts the Pair node created as a positive: their earliest pair (lowest
id, either role, archived included) has them as the positive and its `created_at` is
within 2 s of theirs. A prompt first saved by the PromptManager node, or first as a
negative, is out of scope and not counted.
- Sources: the prompt's non-archived pairs where it is the positive; pairs with a blank
category or no tags are ignored for that field. Category values are trimmed and
case-sensitive, tag sets compared by tag id; a field fills when exactly one value is
left. A whitespace-only category counts as empty.
- Kept, per field, in this order: archived prompt, edited prompt (`updated_at` not within
2 s of `created_at`; NULL or unreadable counts as edited), pairs that disagree, a tag
set over 20 tags. A prompt can fill one field and keep the other.
- The plan lists protected rows first, then fills, then a "Record" row, within the 50
detail rows; at most 20,000 prompts per plan ("review again for the rest").
- The fingerprint covers each considered prompt's computed result (values, tag ids and
names, source pair ids), never a pair's title, notes, rating, run count or use
stamps, so a Pair run between review and apply still applies.
- The apply's writes are guarded (`category` still blank, no `prompt_tags` row) and a
guard that hits nothing rolls the whole transaction back as `plan-changed`. It never
writes `prompt_pairs`, `prompt_pair_tags`, `updated_at`, a tag or Library state.
- The job lists 200 changed rows ("Prompt 12: category “night”; tags a, b"); its
Download report lists every one (`truncated: false`, `Job.record_counted(complete=True)`).
Protected lists one row per kept prompt in the review's wording and order ("Prompt 7:
kept, pairs disagree on category: Adult, asian"), also complete in the report.
- Activity shows a row with no path as a record: its detail as plain text, never an open
or locate action, and the filter reads the detail. A maintenance job of records (every
operation but Missing-file links) says "No records in this group.", "No listed record
matches this filter." and "Filter records…" (`groupCopy`); file jobs keep "files".
Tests: `tests/test_maintenance_pair_metadata.py`, `tests/test_api_maintenance_pair_metadata.py`,
`tests/test_jobs.py`, `tests/js/workspace-system-maintenance.test.js`,
`tests/browser/workspace-system-maintenance.spec.mjs` (hook `/__test__/system/pair-metadata`).
A Python change: ComfyUI needs a restart.
The legacy restore (`POST /prompt_manager/restore`) takes a backup of any size the disks
have room for, so a real library's safety backup restores: the upload, its staged copy
and the swap (the upload twice in the temporary folder, once more plus the current
+17 -4
View File
@@ -280,11 +280,11 @@ def _teardown_logger():
def _stop_background_work():
"""Stop the image monitor, the preview worker and the prompt cleanup
thread against one TEARDOWN_SECONDS deadline, once.
"""Stop the image monitor, the preview worker, the job worker threads
and the prompt cleanup thread against one TEARDOWN_SECONDS deadline, once.
Returns ``{"monitor", "thumbnails", "tracker"}`` booleans (True: it
stopped), leaving out parts that never started. Later calls return a
Returns ``{"monitor", "thumbnails", "workers", "tracker"}`` booleans
(True: it stopped), leaving out parts that never started. Later calls return a
copy of the first result. A Ctrl+C inside a step is logged and the
step is False; one between steps is re-raised after the result (parts
not reached are False) is stored and the ``atexit`` entry removed.
@@ -298,11 +298,14 @@ def _stop_background_work():
api,
_global_prompt_tracker,
)
from .utils import parallel
# A part whose global is set has started; False until its step says
# otherwise, so an interrupt between steps leaves an honest result.
started = (
("monitor", monitor),
("thumbnails", thumbnails),
("workers", parallel.job_pool_started() or None),
("tracker", tracker),
)
result = {name: False for name, part in started if part is not None}
@@ -345,6 +348,16 @@ def _stop_background_work():
result["thumbnails"] = _teardown_step(
"the preview worker", close_previews, logger
)
if "workers" in result:
def stop_workers():
from .utils.parallel import JOB_POOL_JOIN_SECONDS
return parallel.shutdown_job_pool(left(JOB_POOL_JOIN_SECONDS))
result["workers"] = _teardown_step(
"the job worker threads", stop_workers, logger
)
if tracker is not None:
def stop_tracker():
+18 -3
View File
@@ -199,6 +199,23 @@ def link_states(rows, roots, readable):
check's ``missing_files`` and the maintenance missing-link plan both
judge links this way, so their counts agree.
"""
for row, owner, state in _placed_links(rows, roots, readable):
if state is None:
state = "present" if os.path.exists(row[1]) else "missing"
yield row, owner, state
def held_links(rows, roots, readable):
"""``(row, owner, state)`` for each link ``link_states`` holds back
(``unavailable`` or ``outside``), without looking at any file (#586)."""
for row, owner, state in _placed_links(rows, roots, readable):
if state is not None:
yield row, owner, state
def _placed_links(rows, roots, readable):
"""``link_states`` before any file is looked at: state None means the
link lies under a readable source."""
prefixes = [_root_prefix(root) for root in roots]
for row in rows:
owner = _image_owner(row[1], row[3], prefixes)
@@ -206,10 +223,8 @@ def link_states(rows, roots, readable):
yield row, None, "outside"
elif not readable[owner]:
yield row, owner, "unavailable"
elif os.path.exists(row[1]):
yield row, owner, "present"
else:
yield row, owner, "missing"
yield row, owner, None
def _unavailable_line(unavailable, *, partial):
+95 -24
View File
@@ -2,7 +2,8 @@
Four operations, each the native owner of a legacy ``/maintenance`` one and
built on the same queries, so a preview counts what the legacy cleanup would
change:
change (``workflow_storage`` and #550's ``pair_metadata``, which fills old
Pair prompts' empty category and tags, live in their own modules):
- ``duplicate_prompts``: prompts whose text is equal ignoring case and outer
spaces (``DUPLICATE_PROMPTS_SQL``, legacy ``cleanup_duplicates``); each
@@ -42,7 +43,7 @@ from dataclasses import dataclass, field
from pathlib import Path
from .execution_contexts import LIVE_HINT_CTES
from .health import BATCH_ROWS, link_states, readable_roots, readonly_uri
from .health import BATCH_ROWS, held_links, link_states, readable_roots, readonly_uri
from .image_removal_schema import TABLE as MARKS_TABLE
from .image_removal_schema import clear_marks, has_marks_table
from .pairs import PairDependencyError
@@ -60,6 +61,9 @@ MAINTENANCE_OPERATIONS = (
"missing_links",
# Applied as its own batched job (database/workflow_storage.py).
"workflow_storage",
# #550: fill old Pair prompts' empty category and tags
# (database/pair_metadata_repair.py).
"pair_metadata",
)
MAX_DETAILS = 50 # detail rows a plan lists
MAX_RESULT_ROWS = 200 # changed records an apply describes (the job lists 200)
@@ -126,6 +130,16 @@ _UNLINKED_PROTECTIONS_SQL = (
+ "), 0) FROM prompts p WHERE NOT EXISTS "
"(SELECT 1 FROM generated_images gi WHERE gi.prompt_id = p.id)"
)
# Each prompt no image links to, with which protections keep it (#586); the
# ones with none are filtered out in Python, so each predicate runs once.
_UNLINKED_KEPT_SQL = (
"WITH "
+ LIVE_HINT_CTES
+ " SELECT p.id, "
+ ", ".join(f"({predicate})" for predicate, _many, _one in PROTECTIONS)
+ " FROM prompts p WHERE NOT EXISTS "
"(SELECT 1 FROM generated_images gi WHERE gi.prompt_id = p.id) ORDER BY p.id"
)
# Duplicate prompt groups, as the legacy cleanup finds them (case-insensitive
# text; GROUP_CONCAT order is sorted by created_at afterwards).
@@ -155,6 +169,14 @@ _PROMPT_STATE_SQL = (
"ORDER BY id",
)
# Why the missing-link cleanup holds a link back: (one, many) (#586).
HELD_LINK_REASONS = {
"unavailable": (
"link under an unavailable source",
"links under unavailable sources",
),
"outside": ("link outside the image sources", "links outside the image sources"),
}
NO_FILES_DELETED = "No image file is deleted."
LINKS_ONLY = "Database links only; no image file is deleted."
UNUSED_DETAIL = "No image, saved pair, Library state or recorded run uses it"
@@ -168,6 +190,7 @@ CONFLICT_REASONS = {
"groups have their own Library titles",
),
}
PAIRED_DETAIL = "Kept: belongs to a saved pair"
CONFLICT_DETAILS = {
"collision": "Kept: two copies link the same image",
"title": "Kept: a copy has its own Library title",
@@ -409,6 +432,35 @@ def _source_name(root, index):
return os.path.basename(str(root).rstrip("\\/")) or f"Source {index + 1}"
def _kept_duplicates(paired, conflicted):
"""``{prompt_id, reasons}`` for each prompt of a group kept whole (#586).
The reasons are the review's details without "Kept: ".
"""
held = [(ids, PAIRED_DETAIL) for ids in paired] + [
(ids, CONFLICT_DETAILS[reason]) for ids, reason in conflicted
]
return [
{"prompt_id": prompt_id, "reasons": [detail.removeprefix("Kept: ")]}
for ids, detail in held
for prompt_id in ids
]
def _kept_unused(conn):
"""``{prompt_id, reasons}`` for each prompt the unused prune keeps (#586).
The reasons are the review's, one per protection that applies, in the
order the plan names them.
"""
kept = []
for row in conn.execute(_UNLINKED_KEPT_SQL):
reasons = [one for flag, (_p, _m, one) in zip(row[1:], PROTECTIONS) if flag]
if reasons:
kept.append({"prompt_id": row[0], "reasons": reasons})
return kept
class MaintenanceOperations:
"""Mixin for PromptDatabase: maintenance previews and their apply."""
@@ -533,6 +585,10 @@ class MaintenanceOperations:
lambda conn: (self._link_rows(conn), _mark_rows(conn)),
)
return self._plan_missing_links(rows, roots, _stale_marks(marks, roots))
if operation == "pair_metadata":
from .pair_metadata_repair import plan_pair_metadata
return _read_snapshot(self.model.db_path, plan_pair_metadata)
reader = (
self._plan_duplicate_prompts
if operation == "duplicate_prompts"
@@ -584,7 +640,7 @@ class MaintenanceOperations:
+ [
{
"label": _excerpt(texts.get(group[0])),
"detail": "Kept: belongs to a saved pair",
"detail": PAIRED_DETAIL,
}
for group in paired
]
@@ -662,6 +718,24 @@ class MaintenanceOperations:
],
)
def kept_links(self, roots):
"""``{image_path, owner, reason}`` for each link the missing-link
cleanup holds back under ``roots`` (blocking, reads only, no file
opened); ``owner`` indexes ``roots`` (None outside every source).
``reason`` is the review's (``HELD_LINK_REASONS``), one link each.
"""
roots = [str(root) for root in roots]
rows = _read_snapshot(self.model.db_path, self._link_rows)
return [
{
"image_path": row[1],
"owner": owner,
"reason": HELD_LINK_REASONS[state][0],
}
for row, owner, state in held_links(rows, roots, readable_roots(roots))
]
@staticmethod
def _links_fingerprint(candidates, stale=()):
fingerprint = _Fingerprint("missing_links")
@@ -681,23 +755,11 @@ class MaintenanceOperations:
unavailable += 1
elif state == "outside":
outside += 1
reasons = []
if unavailable:
reasons.append(
_plural(
unavailable,
"link under an unavailable source",
"links under unavailable sources",
)
)
if outside:
reasons.append(
_plural(
outside,
"link outside the image sources",
"links outside the image sources",
)
)
reasons = [
_plural(count, *HELD_LINK_REASONS[state])
for state, count in (("unavailable", unavailable), ("outside", outside))
if count
]
count = len(candidates)
parts = []
if count:
@@ -800,7 +862,11 @@ class MaintenanceOperations:
records changed and up to ``MAX_RESULT_ROWS`` descriptions of them
(``{prompt_id[, kept_id]}`` or ``{image_path}``). ``marks`` are a
missing-links plan's stale removal marks; that apply also returns
``marks``, how many it forgot.
``marks``, how many it forgot. ``pair_metadata`` returns every filled
prompt, uncapped (its job keeps the complete record, #550). The
prompt operations also return ``kept``: every prompt the plan keeps,
``{prompt_id, reasons}`` in the review's wording (#550, #586); the
links a missing-links plan keeps are read by ``kept_links``.
"""
if operation not in MAINTENANCE_OPERATIONS or operation == "workflow_storage":
raise ValueError(f"Unknown maintenance operation {operation!r}")
@@ -815,6 +881,10 @@ class MaintenanceOperations:
"unused_components": self._apply_unused_components,
}
with self._pair_transaction() as conn:
if operation == "pair_metadata":
from .pair_metadata_repair import apply_pair_metadata
return apply_pair_metadata(conn, fingerprint, targets)
if operation == "missing_links":
return self._apply_missing_links(
conn, fingerprint, targets, roots, list(marks)
@@ -822,10 +892,11 @@ class MaintenanceOperations:
return appliers[operation](conn, fingerprint, targets, roots)
def _apply_duplicate_prompts(self, conn, fingerprint, targets, _roots):
groups, _paired, _conflicted, current = self._duplicates_fingerprint(conn)
groups, paired, conflicted, current = self._duplicates_fingerprint(conn)
if current != fingerprint or groups != targets:
raise PlanChanged("duplicate_prompts")
changed, rows = 0, []
kept = _kept_duplicates(paired, conflicted)
for sorted_ids in groups:
removed, _moved = self._merge_duplicate_group(conn, sorted_ids, strict=True)
changed += removed
@@ -833,7 +904,7 @@ class MaintenanceOperations:
{"prompt_id": duplicate_id, "kept_id": sorted_ids[0]}
for duplicate_id in sorted_ids[1:]
)
return {"changed": changed, "rows": rows[:MAX_RESULT_ROWS]}
return {"changed": changed, "rows": rows[:MAX_RESULT_ROWS], "kept": kept}
def _apply_unused_components(self, conn, fingerprint, targets, _roots):
ids, current = self._unused_fingerprint(conn)
@@ -845,7 +916,7 @@ class MaintenanceOperations:
{"prompt_id": prompt_id, "text": _excerpt(texts.get(prompt_id))}
for prompt_id in ids[:MAX_RESULT_ROWS]
]
return {"changed": changed, "rows": rows}
return {"changed": changed, "rows": rows, "kept": _kept_unused(conn)}
def _apply_missing_links(self, conn, fingerprint, targets, roots, marks):
rows = []
+452
View File
@@ -0,0 +1,452 @@
"""Pair category and tags repair (#550): the ``pair_metadata`` maintenance plan.
Before #546 a Pair node run saved its category and tags on the saved pair
only; the positive prompt it created stayed Uncategorized and untagged. This
plan fills the **empty** category and the **empty** tag set of those prompts
from their saved pairs (``docs/egm/post-3.4.1/550-design.md``):
- Considered: prompts the Pair node created as a positive (their earliest
pair, in either role, has them as the positive and its ``created_at`` is
within ``PAIR_CREATED_TOLERANCE`` of theirs) whose category is blank
and/or that have no tags.
- Sources: the prompt's non-archived pairs where it is the positive. Pairs
with a blank category or no tags are ignored for that field. A field fills
when exactly one value remains (category: trimmed, case-sensitive; tags:
the tag-id set).
- Kept (protected), in this order: archived prompt, edited prompt
(``updated_at`` not within the tolerance of ``created_at``), pairs that
disagree, a tag set over ``MAX_LIBRARY_TAGS``.
The apply runs in the caller's ``BEGIN IMMEDIATE`` transaction, recomputes
the fingerprint, then writes guarded statements (the field must still be
empty) that never touch a pair, a negative or ``updated_at``. A guard that
hits nothing raises ``PlanChanged`` so the whole transaction rolls back.
"""
import datetime
from . import maintenance as _maintenance
from .library_commands import MAX_LIBRARY_TAGS
from .maintenance import (
MAX_DETAILS,
MaintenanceDraft,
PlanChanged,
_excerpt,
_Fingerprint,
_marks,
_plural,
)
OPERATION = "pair_metadata"
# Seconds: the Pair save stamps its new prompt and its pair with separate
# utc_now_iso() calls in one transaction (microseconds apart).
PAIR_CREATED_TOLERANCE = 2.0
# Fill prompts one plan applies; the rest are applied by the next review.
# Bounds the complete record a job keeps in memory (about 100 bytes a row).
MAX_PLAN_FILLS = 20000
DETAIL_VALUES = 3
DETAIL_TAGS = 5
MEDIA_LABEL = "No image file or saved pair is changed."
EMPTY_LABEL = "No Pair prompts are missing a category or tags"
RECORD_NOTE = (
"The job's report lists every filled field. Download it to keep it: "
"Activity forgets jobs after a restart."
)
# Protected classes in precedence order: (key, detail, reason one, many).
EDITED_DETAIL = (
"Kept: edited after it was saved; its empty category or tags may be intentional"
)
PROTECTED = (
(
"archived",
"Kept: archived in the Library",
"archived in the Library",
"archived in the Library",
),
(
"edited",
EDITED_DETAIL,
"edited after it was saved",
"edited after they were saved",
),
("conflict", None, "where pairs disagree", "where pairs disagree"),
(
"too-many-tags",
f"Kept: its pair has more than {MAX_LIBRARY_TAGS} tags",
f"whose pair has more than {MAX_LIBRARY_TAGS} tags",
f"whose pair has more than {MAX_LIBRARY_TAGS} tags",
),
)
_PROTECTED_ORDER = [key for key, *_rest in PROTECTED]
# Detail rows: what the user can act on (a conflict) first.
DETAIL_ORDER = ("conflict", "edited", "archived", "too-many-tags")
_CANDIDATES_SQL = """
SELECT p.id, p.hash, p.text, p.created_at, p.updated_at,
COALESCE(TRIM(p.category), '') = '' AS blank_category,
NOT EXISTS (SELECT 1 FROM prompt_tags pt WHERE pt.prompt_id = p.id)
AS untagged,
EXISTS (SELECT 1 FROM prompt_archive pa WHERE pa.prompt_id = p.id)
AS archived,
(SELECT CASE WHEN pp.positive_prompt_id = p.id
THEN pp.created_at END
FROM prompt_pairs pp
WHERE pp.positive_prompt_id = p.id OR pp.negative_prompt_id = p.id
ORDER BY pp.id LIMIT 1) AS first_pair_at
FROM prompts p
WHERE EXISTS (SELECT 1 FROM prompt_pairs pp WHERE pp.positive_prompt_id = p.id)
AND (COALESCE(TRIM(p.category), '') = ''
OR NOT EXISTS (SELECT 1 FROM prompt_tags pt WHERE pt.prompt_id = p.id))
ORDER BY p.id
"""
_SOURCES_SQL = (
"SELECT id, positive_prompt_id, TRIM(COALESCE(category, '')) FROM prompt_pairs "
"WHERE archived = 0 AND positive_prompt_id IN ({marks}) ORDER BY id"
)
_PAIR_TAGS_SQL = (
"SELECT ppt.pair_id, t.id, t.name FROM prompt_pair_tags ppt "
"JOIN tags t ON t.id = ppt.tag_id WHERE ppt.pair_id IN ({marks}) "
"ORDER BY ppt.pair_id, t.id"
)
def parse_stamp(value):
"""A stored timestamp as an aware UTC datetime, or None when unreadable."""
if not isinstance(value, str) or not value.strip():
return None
try:
stamp = datetime.datetime.fromisoformat(value.strip())
except ValueError:
return None
if stamp.tzinfo is None:
stamp = stamp.replace(tzinfo=datetime.timezone.utc)
return stamp
def _close(first, second):
"""Whether two stored stamps parse and lie within the tolerance."""
a, b = parse_stamp(first), parse_stamp(second)
if a is None or b is None:
return False
return abs((a - b).total_seconds()) <= PAIR_CREATED_TOLERANCE
def _batches(ids):
size = _maintenance.ID_BATCH
for start in range(0, len(ids), size):
yield ids[start : start + size]
def _sources(conn, ids):
"""``{prompt_id: [(pair_id, category, ((tag_id, name), ...)), ...]}``."""
pairs = {}
for batch in _batches(ids):
for pair_id, prompt_id, category in conn.execute(
_SOURCES_SQL.format(marks=_marks(batch)), batch
):
pairs.setdefault(prompt_id, []).append([pair_id, category, []])
by_pair = {pair[0]: pair for group in pairs.values() for pair in group}
pair_ids = sorted(by_pair)
for batch in _batches(pair_ids):
for pair_id, tag_id, name in conn.execute(
_PAIR_TAGS_SQL.format(marks=_marks(batch)), batch
):
by_pair[pair_id][2].append((tag_id, name))
return {
prompt_id: [(pid, category, tuple(tags)) for pid, category, tags in group]
for prompt_id, group in pairs.items()
}
def _field(values, guard):
"""``(class, value)`` of one field from its distinct values, or None."""
if not values:
return None
if guard:
return (guard, None)
if len(values) > 1:
return ("conflict", sorted(values))
return ("fill", next(iter(values)))
def _classify(row, sources):
"""The result of one considered prompt, or None when nothing applies."""
guard = None
if row["archived"]:
guard = "archived"
elif not _close(row["updated_at"], row["created_at"]):
guard = "edited"
category = tags = None
tag_pair = None
if row["blank_category"]:
category = _field({c for _p, c, _t in sources if c}, guard)
if row["untagged"]:
sets = {}
for pair_id, _category, pair_tags in sources:
if pair_tags:
sets.setdefault(frozenset(pair_tags), pair_id)
tags = _field(set(sets), guard)
if tags and tags[0] == "fill":
tag_pair = sets[tags[1]]
if len(tags[1]) > MAX_LIBRARY_TAGS:
tags = ("too-many-tags", None)
else:
tags = ("fill", tuple(sorted(tags[1])))
elif tags and tags[0] == "conflict":
tags = ("conflict", [sorted(s) for s in sorted(sets, key=sorted)])
if category is None and tags is None:
return None
return {
"id": row["id"],
"hash": row["hash"],
"text": row["text"],
"category": category,
"tags": tags,
"tag_pair": tag_pair if tags and tags[0] == "fill" else None,
"pairs": [pair_id for pair_id, _c, _t in sources],
}
def _state(conn):
"""``(results, fingerprint)``: every considered prompt's result, by id."""
considered = []
for row in conn.execute(_CANDIDATES_SQL):
row = dict(
zip(
(
"id",
"hash",
"text",
"created_at",
"updated_at",
"blank_category",
"untagged",
"archived",
"first_pair_at",
),
tuple(row),
)
)
if _close(row["created_at"], row["first_pair_at"]):
considered.append(row)
sources = _sources(conn, [row["id"] for row in considered])
fingerprint = _Fingerprint(OPERATION)
results = []
for row in considered:
result = _classify(row, sources.get(row["id"], []))
if result is None:
continue
results.append(result)
fingerprint.add(
[
result["id"],
result["hash"],
result["category"],
result["tags"],
result["tag_pair"],
result["pairs"],
]
)
return results, fingerprint.hexdigest()
def _fills(result):
return (
bool(result["category"] and result["category"][0] == "fill"),
bool(result["tags"] and result["tags"][0] == "fill"),
)
def _targets(results):
targets = []
for result in results:
category, tags = _fills(result)
if category or tags:
targets.append([result["id"], category, tags])
return targets[:MAX_PLAN_FILLS]
def _protected_classes(result):
"""The protected classes of a result's fields, in precedence order."""
found = {
field[0]
for field in (result["category"], result["tags"])
if field and field[0] != "fill"
}
return [key for key in _PROTECTED_ORDER if key in found]
def _more(values, limit):
shown = ", ".join(values[:limit])
return shown + (", …" if len(values) > limit else "")
def _fill_detail(result):
parts = []
category, tags = _fills(result)
if category:
parts.append(f"category “{result['category'][1]}”")
if tags:
names = sorted(name for _id, name in result["tags"][1])
noun = "tag" if len(names) == 1 else "tags"
parts.append(f"{len(names)} {noun}: {_more(names, DETAIL_TAGS)}")
pairs = [str(pair_id) for pair_id in result["pairs"]]
label = "pair" if len(pairs) == 1 else "pairs"
return f"Fills {' and '.join(parts)} ({label} {_more(pairs, DETAIL_VALUES)})"
def _protected_detail(result, key):
if key != "conflict":
return next(detail for k, detail, *_r in PROTECTED if k == key)
if result["category"] and result["category"][0] == "conflict":
values = result["category"][1]
return f"Kept: pairs disagree on category: {_more(values, DETAIL_VALUES)}"
return "Kept: pairs disagree on tags"
def _kept(results):
"""One row per kept prompt, its review reasons without "Kept: " (#550)."""
rows = []
for result in results:
classes = set(_protected_classes(result))
if classes:
rows.append(
{
"prompt_id": result["id"],
"reasons": [
_protected_detail(result, key).removeprefix("Kept: ")
for key in DETAIL_ORDER
if key in classes
],
}
)
return rows
def _details(results, targets):
rows = []
for key in DETAIL_ORDER:
for result in results:
if key in _protected_classes(result):
rows.append(
{
"label": _excerpt(result["text"]),
"detail": _protected_detail(result, key),
}
)
planned = {target[0] for target in targets}
for result in results:
if result["id"] in planned:
rows.append(
{"label": _excerpt(result["text"]), "detail": _fill_detail(result)}
)
if not targets:
return rows[:MAX_DETAILS]
return rows[: MAX_DETAILS - 1] + [{"label": "Record", "detail": RECORD_NOTE}]
def _scope_label(targets, total):
if not targets:
return EMPTY_LABEL
categories = sum(1 for _id, category, _tags in targets if category)
tag_sets = sum(1 for _id, _category, tags in targets if tags)
label = (
f"{_plural(len(targets), 'prompt')} to fill: "
f"{_plural(categories, 'category', 'categories')}, "
f"{_plural(tag_sets, 'tag set')}"
)
if total > len(targets):
label += (
f" (the first {len(targets):,} of {total:,}; review again for the rest)"
)
return label
def plan_pair_metadata(conn):
"""The reviewed ``pair_metadata`` plan on ``conn``'s snapshot."""
results, fingerprint = _state(conn)
targets = _targets(results)
total = sum(1 for result in results if any(_fills(result)))
counts = {key: 0 for key in _PROTECTED_ORDER}
held = 0
for result in results:
classes = _protected_classes(result)
held += bool(classes)
for key in classes:
counts[key] += 1
reasons = [
f"{counts[key]:,} {one if counts[key] == 1 else many}"
for key, _detail, one, many in PROTECTED
if counts[key]
]
return MaintenanceDraft(
operation=OPERATION,
fingerprint=fingerprint,
targets=targets,
scope={"count": len(targets), "label": _scope_label(targets, total)},
protected={"count": held, "reasons": reasons},
media={"deletes_files": False, "label": MEDIA_LABEL},
details=_details(results, targets),
)
def _fill_categories(conn, fills):
"""Set each blank category; raises PlanChanged when one is no longer blank."""
cursor = conn.executemany(
"UPDATE prompts SET category = ? WHERE id = ? "
"AND COALESCE(TRIM(category), '') = ''",
fills,
)
if cursor.rowcount != len(fills):
raise PlanChanged(OPERATION)
def _fill_tags(conn, fills, expected):
"""Copy each source pair's tags onto its untagged prompt."""
cursor = conn.executemany(
"INSERT INTO prompt_tags(prompt_id, tag_id) "
"SELECT ?, ppt.tag_id FROM prompt_pair_tags ppt "
"JOIN tags t ON t.id = ppt.tag_id WHERE ppt.pair_id = ? "
"AND NOT EXISTS (SELECT 1 FROM prompt_tags WHERE prompt_id = ?)",
fills,
)
if cursor.rowcount != expected:
raise PlanChanged(OPERATION)
def apply_pair_metadata(conn, fingerprint, targets):
"""Apply the reviewed fills on ``conn`` (inside BEGIN IMMEDIATE).
Returns ``{changed, categories, tag_sets, rows, kept}``; ``rows`` lists
every filled prompt (``{prompt_id, category?, tags?}``) and ``kept`` every
prompt the plan keeps (``{prompt_id, reasons}``, in the review's wording
and order), both never capped.
"""
results, current = _state(conn)
if current != fingerprint or _targets(results) != targets:
raise PlanChanged(OPERATION)
by_id = {result["id"]: result for result in results}
categories, tags, rows = [], [], []
tag_count = 0
for prompt_id, fill_category, fill_tags in targets:
result = by_id[prompt_id]
row = {"prompt_id": prompt_id}
if fill_category:
row["category"] = result["category"][1]
categories.append((row["category"], prompt_id))
if fill_tags:
row["tags"] = sorted(name for _id, name in result["tags"][1])
tags.append((prompt_id, result["tag_pair"], prompt_id))
tag_count += len(result["tags"][1])
rows.append(row)
if categories:
_fill_categories(conn, categories)
if tags:
_fill_tags(conn, tags, tag_count)
return {
"changed": len(rows),
"categories": len(categories),
"tag_sets": len(tags),
"rows": rows,
"kept": _kept(results),
}
+16 -11
View File
@@ -112,20 +112,22 @@ QUERY_PLAN_STATEMENTS = {
def _rewrite_row(conn, image_id, row, verified, checked):
"""Store the row's inline texts as documents; returns (changed, protected, size).
"""Store the row's inline texts as documents; returns (changed, kept, size).
``kept`` names the columns whose value stays inline (not text).
``verified`` maps document ids already read back in this batch to their
text, and ``checked`` holds the reused ids ``store_text`` already
checked, so a document shared by many rows is decompressed once.
"""
changes, replaced = {}, []
protected = size = 0
changes, replaced, kept = {}, [], []
size = 0
for index, (column, reference) in enumerate(TEXT_COLUMNS):
value, current = row[index], row[index + len(TEXT_COLUMNS)]
if value is None:
continue
if not storable(value):
protected += 1
kept.append(column)
continue
document_id = store_text(conn, value, checked)
if document_id not in verified:
@@ -143,7 +145,7 @@ def _rewrite_row(conn, image_id, row, verified, checked):
[*changes.values(), image_id],
)
release_documents(conn, replaced)
return bool(changes), protected, size
return bool(changes), kept, size
def migrate_batch(
@@ -160,9 +162,10 @@ def migrate_batch(
savepoint: a row that cannot be rewritten (``_row_error``) is rolled
back alone, stays inline and is listed in ``errors``; anything else rolls
the whole batch back and propagates. Returns ``{last_id, rows,
protected, errors, done}``: the last row looked at, rows rewritten,
inline values kept (not text), the ids kept by an error and whether
there was nothing left to look at.
protected, kept, errors, done}``: the last row looked at, rows
rewritten, inline values kept (not text) and each of them as ``(image
id, column)`` (#586), the ids kept by an error and whether there was
nothing left to look at.
"""
if conn.in_transaction:
raise RuntimeError("The migration needs its own transaction")
@@ -174,6 +177,7 @@ def migrate_batch(
"last_id": after_id,
"rows": 0,
"protected": 0,
"kept": [],
"errors": [],
"done": not ids,
}
@@ -182,7 +186,7 @@ def migrate_batch(
conn.execute("SAVEPOINT workflow_row")
try:
row = conn.execute(ROW_SQL, (image_id,)).fetchone()
changed, protected, row_size = _rewrite_row(
changed, kept, row_size = _rewrite_row(
conn, image_id, row, verified, checked
)
except Exception as error:
@@ -194,11 +198,12 @@ def migrate_batch(
checked.clear()
logger.warning(f"Image record {image_id} kept inline: {error}")
result["errors"].append(image_id)
changed, protected, row_size = False, 0, 0
changed, kept, row_size = False, [], 0
else:
conn.execute("RELEASE workflow_row")
result["rows"] += int(changed)
result["protected"] += protected
result["protected"] += len(kept)
result["kept"].extend((image_id, column) for column in kept)
result["last_id"] = image_id
size += row_size
if size >= max_bytes or _clock() - started >= max_seconds:
+15 -1
View File
@@ -85,6 +85,8 @@ try:
from ...database.operations import PromptDatabase
from ...utils.file_prompt import parse_comfyui_prompt
from ...utils.logging_config import get_logger
from ...utils.native_memory import release_free_memory
from ...utils.png_text import png_text
from ...utils.video_metadata import is_video_path, read_video_metadata
except ImportError:
import sys
@@ -95,6 +97,8 @@ except ImportError:
from database.operations import PromptDatabase
from utils.file_prompt import parse_comfyui_prompt
from utils.logging_config import get_logger
from utils.native_memory import release_free_memory
from utils.png_text import png_text
from utils.video_metadata import is_video_path, read_video_metadata
@@ -526,6 +530,14 @@ class PromptManagerAPI(
return await loop.run_in_executor(None, call)
return await loop.run_in_executor(None, func, *args)
async def _release_job_memory(self):
"""After a scan or thumbnail job: glibc's free heap pages go back to
the OS once (``malloc_trim(0)``; nothing elsewhere, #523)."""
try:
await self._run_in_executor(release_free_memory)
except Exception: # best effort: the job's result already stands
self.logger.debug("Releasing free memory failed", exc_info=True)
def _root_listing(self, root) -> RootListing:
"""The shared listing record of one output root (created empty)."""
return self._root_listings.setdefault(str(root), RootListing())
@@ -1048,6 +1060,8 @@ class PromptManagerAPI(
def _extract_comfyui_metadata(self, image_path):
"""Embedded ComfyUI metadata of an image (PNG text chunks) or a video.
PNG text is read without decoding the pixels (``utils.png_text``, #523).
Videos go through ffprobe (see utils/video_metadata). Unreadable files
are a warning, not an error: a scan over a large output tree meets
truncated and foreign files, and it carries on past them.
@@ -1056,7 +1070,7 @@ class PromptManagerAPI(
return read_video_metadata(image_path)
try:
with Image.open(image_path) as img:
return dict(getattr(img, "text", None) or {})
return png_text(img)
except Exception as e:
self.logger.warning(
f"Unreadable image {os.path.basename(str(image_path))}: {e}"
+5 -3
View File
@@ -1454,6 +1454,7 @@ class AdminRoutesMixin:
verb = "cancelled" if state == "cancelled" else "finished"
job.log_line(f"Scan {verb} · {_plural(job.processed, 'file')} processed")
job.finish(state)
await self._release_job_memory()
if state == "finished":
# After the scan, so the guarded thumbnail job is admitted (#249).
await self._hand_off_scan_thumbnails(job, output_dirs)
@@ -1668,9 +1669,10 @@ class AdminRoutesMixin:
def _extract_batch_metadata_sync(self, file_batch):
"""Extract metadata from a batch of files in one executor call.
Files are read on ``PromptManagerConfig.WORKER_THREADS`` threads:
Pillow decodes the whole image to reach trailing text chunks and
releases the GIL while doing so.
Files are read on the long-lived job pool of
``PromptManagerConfig.WORKER_THREADS`` threads (#523).
PNG text is read chunk by chunk without decoding pixels (#523); file
reads and ffprobe release the GIL.
"""
from ..config import PromptManagerConfig
+12 -1
View File
@@ -65,6 +65,7 @@ from .duplicate_review import ( # noqa: F401 (CHANGED, OUTSIDE_SOURCE: public)
NOT_ATTEMPTED,
OUTSIDE_SOURCE,
RECORDS_FAILED,
REVIEW_KEPT_DETAILS,
ROW_DETAILS,
SAME_FILE,
SOURCES_CHANGED,
@@ -434,6 +435,7 @@ class DuplicateFilesMixin:
{
"targets": review["targets"],
"kept": review["kept"],
"held": review["held"],
"protected": review["public"]["protected"]["count"],
},
)
@@ -446,7 +448,7 @@ class DuplicateFilesMixin:
job hashes each file again right before it unlinks it.
"""
listed = self._listed_sources()
targets, kept, protected = [], [], {}
targets, kept, protected, held = [], [], {}, []
for group, keepers in selection.values():
removals = [r for r in group["files"] if r not in keepers]
for record in removals:
@@ -467,6 +469,7 @@ class DuplicateFilesMixin:
problem = _bound_problem(record, group, keeper)
if problem:
protected[problem] = protected.get(problem, 0) + 1
held.append(self._copy_row(record, REVIEW_KEPT_DETAILS[problem]))
continue
reviewed.append(self._target(group, record, keeper))
if reviewed:
@@ -488,6 +491,7 @@ class DuplicateFilesMixin:
return {
"targets": targets,
"kept": kept,
"held": held,
"fingerprint": _fingerprint(targets),
"public": _public_plan(targets, protected, records),
}
@@ -584,6 +588,13 @@ class DuplicateFilesMixin:
async def _remove_duplicates(self, job, db, plan, backup):
targets = plan.payload["targets"]
# #586: every copy the review protected, with its reason, and the
# whole group in the report.
held = plan.payload["held"]
job.record_counted("kept", len(held), held, complete=True)
if held:
copies = _plural(len(held), "copy", "copies")
job.log_line(f"Kept {copies} the review protected")
for row in plan.payload["kept"]:
job.record("kept", row, f"= {row['root']}/{row['path']} · {KEPT_COPY}")
job.update(total=len(targets), status_line="Removing duplicate files")
+12
View File
@@ -70,6 +70,18 @@ PROTECTED_REASONS = {
"files kept: their kept copy changed or is gone",
),
}
# The job's row for a copy the review protected (#586): the review's reason.
REVIEW_KEPT_DETAILS = {
MAYBE_SAME: "Kept: may be the same file reached through another mount or share",
NO_FILE_ID: "Kept: can't confirm this is a separate file on this drive",
SAME: "Kept: the same file as its kept copy",
OUTSIDE_SOURCE: "Kept: not inside a configured source",
LINK: "Kept: a link, not a regular file",
CHANGED_CONTENT: "Kept: changed since the search",
GONE: "Kept: no longer exists",
UNREADABLE: "Kept: could not be read",
KEEPER: "Kept: its kept copy changed or is gone",
}
# Job outcome rows (never an OS error text).
CHANGED = "The file changed since the review; it was kept."
KEEPER_CHANGED = "The kept copy changed or is gone; this file was kept."
+4 -1
View File
@@ -843,7 +843,10 @@ class HistoryRoutesMixin:
raise
async def execute():
pending = asyncio.create_task(self._run_in_executor(project))
# An inspection reads the file's header and text: the
# interactive image pool (#523); media stays on the default.
run = self._run_image if kind == "inspection" else self._run_in_executor
pending = asyncio.create_task(run(project))
try:
return await asyncio.shield(pending)
except asyncio.CancelledError:
+31 -11
View File
@@ -2975,6 +2975,7 @@ class ImageRoutesMixin:
result = await self._run_in_executor(
self._generate_gallery_thumbnails_sync, output_dirs
)
await self._release_job_memory()
return web.json_response(result)
except Exception as e:
@@ -3997,7 +3998,8 @@ class ImageRoutesMixin:
**observed,
)
body = await self._run_in_executor(inspect)
# Header and text reads: the interactive image pool (#523).
body = await self._run_image(inspect)
if body is None:
return web.json_response(
{
@@ -4210,6 +4212,22 @@ class ImageRoutesMixin:
# (which Windows refuses on open files) is not held up by it.
return _CardFacts(width, height, *self._card_parse(path, records))
def _cached_card_facts(self, entry: tuple) -> _CardFacts | None:
path, mtime, root, index = entry[:4]
return self._workspace_card_cache.get((index, str(root)), {}).get((path, mtime))
def _card_facts_of(self, entries: list, *, parse: bool) -> list[_CardFacts]:
"""Card facts of *entries*, in order; files not cached yet are read on
the interactive image pool (#523), cached ones answer here."""
found = [self._cached_card_facts(entry) for entry in entries]
missing = [i for i, facts in enumerate(found) if facts is None]
read = self._map_image(
lambda i: self._card_facts(entries[i], parse=parse), missing
)
for i, facts in zip(missing, read):
found[i] = facts
return found
def _card_facts(self, entry: tuple, *, parse: bool) -> _CardFacts:
"""One listed image's card facts, read once per ``(path, mtime)``.
@@ -4218,11 +4236,11 @@ class ImageRoutesMixin:
and drops the rest, and a roots change drops them all. Only files of
the current scan are stored, so the cache is bounded by the listing.
"""
path, mtime, root, index = entry[:4]
root_key, key = (index, str(root)), (path, mtime)
cached = self._workspace_card_cache.get(root_key, {}).get(key)
cached = self._cached_card_facts(entry)
if cached is not None:
return cached
path, mtime, root, index = entry[:4]
root_key, key = (index, str(root)), (path, mtime)
facts = self._read_card_facts(path, parse=parse)
if facts is None:
# Not stored: a locked or half-written file may read next time.
@@ -4248,15 +4266,17 @@ class ImageRoutesMixin:
if linked
else {}
)
# Videos: the card reads no video metadata. Library previews show
# no card facts, so they skip the parse (``parse=linked``).
stills = [image for image in images if not image["is_video"]]
read = self._card_facts_of(
[entries[(image["root_index"], image["path"])] for image in stills],
parse=linked,
)
facts_of = {id(image): facts for image, facts in zip(stills, read)}
for image in images:
entry = entries[(image["root_index"], image["path"])]
# Videos: the card reads no video metadata. Library previews show
# no card facts, so they skip the parse (``parse=linked``).
facts = (
_NO_CARD_FACTS
if image["is_video"]
else self._card_facts(entry, parse=linked)
)
facts = facts_of.get(id(image), _NO_CARD_FACTS)
image["width"], image["height"] = facts.width, facts.height
if not linked:
continue
+39 -25
View File
@@ -204,6 +204,8 @@ class Job:
self.total = 0
self.counters = dict.fromkeys(groups, 0)
self.outcomes = {group: [] for group in groups}
# Groups whose report lists every row, beyond MAX_OUTCOME_ROWS (#550).
self.complete = {}
self.log = []
self._cancel_requested = False
self.last_event = None
@@ -286,42 +288,48 @@ class Job:
index = row.get("root_index")
if isinstance(index, bool) or not isinstance(index, int) or index < 0:
index = None
shaped = {
"path": path,
"root": str(row.get("root") or ""),
"root_index": index,
"detail": str(row.get("detail") or ""),
}
if group in self.complete: # a group kept whole for the report (#586)
self.complete[group].append(dict(shaped))
if len(rows) < MAX_OUTCOME_ROWS:
rows.append(
{
"path": path,
"root": str(row.get("root") or ""),
"root_index": index,
"detail": str(row.get("detail") or ""),
}
)
rows.append(shaped)
self.log_line(log_line)
def record_counted(self, group, count, rows=()):
def record_counted(self, group, count, rows=(), *, complete=False):
"""Count ``count`` outcomes in ``group`` at once, listing ``rows``.
For outcomes that need no log line each (a thumbnail job's already
fresh previews): the counter is exact, ``rows`` (``record``'s row
shape) are listed up to MAX_OUTCOME_ROWS, and nothing is logged.
``complete`` also keeps every row for the report (#550: a record of
each change that must not be cut short; the caller bounds it).
"""
if group not in self.counters:
raise ValueError(f"Unknown outcome group {group!r} for {self.kind}")
self.counters[group] += max(0, int(count))
listed = self.outcomes[group]
kept = self.complete.setdefault(group, []) if complete else None
for row in rows:
if len(listed) >= MAX_OUTCOME_ROWS:
if kept is None and len(listed) >= MAX_OUTCOME_ROWS:
break
index = row.get("root_index")
if isinstance(index, bool) or not isinstance(index, int) or index < 0:
index = None
listed.append(
{
"path": _relative_path(row.get("path")),
"root": str(row.get("root") or ""),
"root_index": index,
"detail": str(row.get("detail") or ""),
}
)
shaped = {
"path": _relative_path(row.get("path")),
"root": str(row.get("root") or ""),
"root_index": index,
"detail": str(row.get("detail") or ""),
}
if kept is not None:
kept.append(shaped)
if len(listed) < MAX_OUTCOME_ROWS:
listed.append(dict(shaped))
def finish(self, state="finished"):
"""End the job in ``state`` and release every subscriber.
@@ -406,8 +414,9 @@ class Job:
"""The downloadable report: the summary, every outcome group and the log.
Each group carries its exact count, its listed rows (at most
MAX_OUTCOME_ROWS) and whether the list is cut short. Rows and log lines
already hold root-relative paths only.
MAX_OUTCOME_ROWS, or every row of a ``complete`` group) and whether
the list is cut short. Rows and log lines already hold root-relative
paths only.
"""
summary = self.summary()
summary.pop("listed", None)
@@ -417,16 +426,21 @@ class Job:
"generated_at": self._clock(),
"job": summary,
"outcomes": {
group: {
"count": self.counters.get(group, len(rows)),
"rows": [dict(row) for row in rows],
"truncated": self.counters.get(group, len(rows)) > len(rows),
}
group: self._report_group(group, rows)
for group, rows in self.outcomes.items()
},
"log": list(self.log),
}
def _report_group(self, group, listed):
rows = self.complete.get(group, listed)
count = self.counters.get(group, len(rows))
return {
"count": count,
"rows": [dict(row) for row in rows],
"truncated": count > len(rows),
}
def report_filename(self):
"""The download name: prompt-manager-KIND-YYYYMMDD-HHMM.json (local time)."""
stamp = time.strftime("%Y%m%d-%H%M", time.localtime(self._clock()))
+23 -2
View File
@@ -68,6 +68,23 @@ def _count(count, singular, plural):
return f"{count} {singular if count == 1 else plural}"
# Each kept entry's row in the job (#586), in the review's "Kept: ..." words.
KEPT_DETAILS = {
"active": "Kept: the active log",
"link": "Kept: a link",
"other": "Kept: another file",
}
def _kept_files(rows):
"""``{name, kind}`` of each entry the cleanup keeps, in folder order."""
return [
{"name": row["name"], "kind": row["kind"]}
for row in rows
if row["kind"] in KEPT_DETAILS
]
def _protected_reasons(rows):
"""The plan's Protected reasons: what the cleanup keeps, zero counts left out."""
active = sum(1 for row in rows if row["kind"] == "active")
@@ -129,7 +146,7 @@ class LogCleanupMixin:
token = store.issue(
ROTATED_LOGS,
json.dumps(files, sort_keys=True),
{"files": files, "protected": protected},
{"files": files, "protected": protected, "kept": _kept_files(rows)},
(),
)
except ScopeTooLarge:
@@ -198,7 +215,11 @@ class LogCleanupMixin:
job.record(
"errors", {"path": name, "detail": detail}, f"! {name} · {detail}"
)
job.record_counted("protected", plan.payload["protected"])
kept = [
{"path": row["name"], "detail": KEPT_DETAILS[row["kind"]]}
for row in plan.payload["kept"]
]
job.record_counted("protected", len(kept), kept, complete=True)
removed, failed = len(result["removed"]), len(result["failed"])
line = f"Deleted {_count(removed, 'rotated log copy', 'rotated log copies')}"
if failed:
+95 -6
View File
@@ -1,8 +1,9 @@
"""Maintenance routes: reviewed, read-only plans (#249 System S6).
``POST /prompt_manager/maintenance/preview {operation, options:{}}`` reviews
one operation (``duplicate_prompts``, ``optimize``, ``unused_components`` or
``missing_links``; ``database/maintenance.py``; ``duplicate_files``,
one operation (``duplicate_prompts``, ``optimize``, ``unused_components``,
``missing_links`` or ``pair_metadata``, #550; ``database/maintenance.py``;
``duplicate_files``,
``py/api/duplicate_files.py``; ``rotated_logs``, the rotated log copies,
``py/api/log_cleanup.py``, which has its own apply job) off the event loop and
answers a ``Plan``: ``{success, token, operation, expires_at, scope:{count,
@@ -97,7 +98,11 @@ MAINTENANCE_TITLES = {
"unused_components": "Unused prompt components",
"missing_links": "Missing-file links",
"workflow_storage": "Store each workflow once",
# #550, beyond the concept: fills old Pair prompts' empty category and tags.
"pair_metadata": "Pair category and tags",
}
# Operations whose job keeps every changed row for its report (#550).
COMPLETE_RECORD = frozenset({"pair_metadata"})
_OPERATION_ERROR = "operation must be one of: " + ", ".join(PREVIEW_OPERATIONS)
MAINTENANCE_GROUPS = ("changed", "protected", "errors")
MAX_TOKEN_LENGTH = 128
@@ -281,9 +286,27 @@ def short_space_message(short):
)
def _changed_line(operation, count, marks=0):
def _pair_metadata_line(result):
count = result["changed"]
noun = (
"prompt from its saved pairs"
if count == 1
else "prompts from their saved pairs"
)
categories = result.get("categories", 0)
tag_sets = result.get("tag_sets", 0)
return (
f"Filled {count:,} {noun} "
f"({categories:,} {'category' if categories == 1 else 'categories'}, "
f"{tag_sets:,} {'tag set' if tag_sets == 1 else 'tag sets'})"
)
def _changed_line(operation, count, marks=0, result=None):
if operation == "optimize":
return "Optimized the database"
if operation == "pair_metadata":
return _pair_metadata_line(result or {"changed": count})
if operation == "duplicate_prompts":
noun = "duplicate prompt" if count == 1 else "duplicate prompts"
return f"Merged {count:,} {noun}"
@@ -668,13 +691,67 @@ class MaintenanceRoutesMixin:
rows = await self._run_in_executor(
self._maintenance_rows, plan.operation, result["rows"]
)
job.record_counted("changed", result["changed"], rows)
job.record_counted("protected", payload["protected"])
line = _changed_line(plan.operation, result["changed"], result.get("marks", 0))
job.record_counted(
"changed",
result["changed"],
rows,
complete=plan.operation in COMPLETE_RECORD,
)
kept = await self._run_in_executor(self._kept_rows, db, plan, result)
if kept is None:
job.record_counted("protected", payload["protected"]) # none kept
else:
# #550, #586: each kept record with its reason, like the changed
# ones, and every one of them in the report.
job.record_counted("protected", len(kept), kept, complete=True)
line = _changed_line(
plan.operation, result["changed"], result.get("marks", 0), result
)
job.log_line(line)
job.update(progress=100, status_line=f"{line} · {backup_note(backup)}")
return None
def _kept_rows(self, db, plan, result):
"""Outcome rows for the records the apply kept (blocking), or None.
Prompts are records (no path); a missing-links plan's kept links
name their file, relative to its output root, so Activity can open
and locate it (#586); one under a source that is gone keeps its path
under that source's label. None for an operation that keeps nothing
(Optimize).
"""
if plan.operation == "missing_links":
roots = _scan_roots(self._get_all_output_dirs())
judged = plan.payload["roots"]
rows = []
for item in db.kept_links(judged):
path, label, index = _scan_location(item["image_path"], roots)
if index is None and item["owner"] is not None:
# A source that is gone is not listed: name the file
# under it, with no listing index to open it by.
source = judged[item["owner"]]
where = [(os.path.abspath(source), self._public_path(source), None)]
path, label, index = _scan_location(item["image_path"], where)
rows.append(
{
"path": path,
"root": label,
"root_index": index,
"detail": f"Kept: {item['reason']}",
}
)
return rows
if "kept" not in result:
return None
return [
{
"path": "",
"detail": f"Prompt {item['prompt_id']}: kept, "
+ "; ".join(item["reasons"]),
}
for item in result["kept"]
]
def _maintenance_rows(self, operation, described):
"""Outcome rows for the changed records (root-relative paths only)."""
if operation == "missing_links":
@@ -691,6 +768,8 @@ class MaintenanceRoutesMixin:
}
)
return rows
if operation == "pair_metadata":
return [{"path": "", "detail": _filled_detail(item)} for item in described]
if operation == "duplicate_prompts":
return [
{
@@ -708,6 +787,16 @@ class MaintenanceRoutesMixin:
]
def _filled_detail(item):
"""``Prompt 12: category “portrait”; tags a, b``: one filled prompt (#550)."""
parts = []
if "category" in item:
parts.append(f"category “{item['category']}”")
if item.get("tags"):
parts.append("tags " + ", ".join(item["tags"]))
return f"Prompt {item['prompt_id']}: " + "; ".join(parts)
def backup_note(backup):
"""The status line's backup part: the file name, or that there is none."""
return f"Safety backup {backup}" if backup else NO_BACKUP
+89 -26
View File
@@ -8,7 +8,8 @@ previews behind those URLs are made here, in three ways:
- **On first view.** ``GET /prompt_manager/images/serve/thumbnails/<rel>_thumb
<ext>`` for a preview that does not exist yet is admitted on the event loop
(``_PreviewGate``): at most ``ON_DEMAND_LIMIT`` previews are written at once,
on this API's own small pool (never the default executor), one flight runs
on this API's interactive image pool (``IMAGE_POOL_SIZE`` threads, shared
with card facts and inspector reads; never the default executor), one flight runs
per file, and the requests waiting are capped (``MAX_WAITING``); past the
cap a request gets its fallback at once. Inside the admission the preview is
mapped back to its media file under the same root (the request text is
@@ -24,7 +25,8 @@ previews behind those URLs are made here, in three ways:
- **On capture.** The image monitor hands every new output it settles to
``queue_capture_thumbnail``, which queues its preview for one background
worker thread (``AutoThumbnailer``). Each wake takes at most Worker threads
(``WORKER_THREADS``) files and writes them on that many threads; enqueuing
(``WORKER_THREADS``) files and writes them on the long-lived job pool
(``utils.parallel``, shared with scans and thumbnail jobs); enqueuing
is O(1), so the monitor and ComfyUI's queue never wait. A file already
queued is not queued again, the queue is bounded (``MAX_QUEUED``) and a
preview left out is made on its first view instead.
@@ -49,7 +51,7 @@ write starts (after its slot) and again just before its rename.
Shutdown: ``close_auto_thumbnails`` stops the worker at a file boundary,
kills a running ffmpeg, waits a bounded time (and logs when the wait ran
out) and shuts the view pool down; after the app's cleanup
out) and shuts the image pool down; after the app's cleanup
(``final=True``) nothing starts a worker again. Importing this module again
(a package reload) stops the workers the previous import left.
"""
@@ -60,7 +62,7 @@ import functools
import io
import os
import threading
from concurrent.futures import ThreadPoolExecutor
from concurrent.futures import CancelledError, ThreadPoolExecutor
from pathlib import Path
from aiohttp import web
@@ -104,8 +106,11 @@ PLACEHOLDER_SIZE = (thumbnail_settings.DEFAULT_EDGE, thumbnail_settings.DEFAULT_
PLACEHOLDER_QUALITY = 85
# Source extensions in the order a shared preview name looks for them (R2).
_SOURCE_ORDER = thumbnail_settings.IMAGE_SOURCE_ORDER + tuple(sorted(VIDEO_EXTENSIONS))
# Previews views may write at once (the size of the view pool).
ON_DEMAND_LIMIT = 3
# Threads of the interactive image pool (#523): view previews, card facts and
# inspector reads. Fixed, so image buffers stay in a few threads' arenas.
IMAGE_POOL_SIZE = 3
# Previews views may write at once (the whole image pool).
ON_DEMAND_LIMIT = IMAGE_POOL_SIZE
# View requests that may wait on the gate; one more gets its fallback at once.
MAX_WAITING = 64
# How long a pool thread waits for the capture worker's flight of the same file.
@@ -124,7 +129,7 @@ AUTO_REFUSED_KINDS = frozenset(
{RESTORE_JOB_KIND, DUPLICATE_CLEANUP_JOB_KIND, SOURCE_CHANGE_JOB_KIND}
)
WORKER_NAME = "PromptManagerThumbnailWorker"
POOL_PREFIX = "pm-preview"
POOL_PREFIX = "pm-image"
# Identifies this import's workers: a reload stops the ones it did not start.
_MODULE_TOKEN = object()
@@ -294,7 +299,7 @@ class AutoThumbnailer:
``generate(src, dst, is_video, *, stop, guard, track)`` writes one
preview and returns ``_generate_one``'s result dict; ``may_write()``
says whether automatic previews may be written now. Views call
``ensure`` on the view pool, the monitor calls ``enqueue`` from its
``ensure`` on the image pool, the monitor calls ``enqueue`` from its
thread.
"""
@@ -491,12 +496,15 @@ class AutoThumbnailer:
self._worker_running = False
self._changed.notify_all()
return
take = min(len(self._queue), max(1, int(self._workers() or 1)))
workers = max(1, int(self._workers() or 1))
take = min(len(self._queue), workers)
batch = [self._queue.popleft() for _ in range(take)]
self._batch = len(batch)
self._changed.notify_all()
try:
map_parallel(self._write_queued, batch, len(batch))
# The job pool's size is the setting; a smaller batch uses
# fewer of its threads (#523: one long-lived pool).
map_parallel(self._write_queued, batch, workers)
finally:
with self._changed:
self._batch = 0
@@ -567,7 +575,7 @@ class _PreviewGate:
"""Admission of view requests on the event loop (#249 I2).
Waiters are coroutines: at most ``limit`` previews are written at once
on the view pool, one flight runs per key, and past ``MAX_WAITING``
on the image pool, one flight runs per key, and past ``MAX_WAITING``
waiting requests ``run`` answers ``OVER`` at once.
"""
@@ -622,13 +630,17 @@ class _PreviewGate:
flight.set_result(result)
def _on_image_thread():
return threading.current_thread().name.startswith(POOL_PREFIX + "_")
class AutoThumbnailMixin:
"""Automatic previews on the API: views, the monitor's queue, scans."""
_auto_thumbnailer = None
_auto_thumbnails_shut = False
_preview_gate_obj = None
_preview_pool_obj = None
_image_pool_obj = None
_auto_thumbnailer_lock = threading.Lock()
def _auto_thumbnails(self):
@@ -651,7 +663,7 @@ class AutoThumbnailMixin:
self._auto_thumbnails_shut = True
def close_auto_thumbnails(self, final=False, timeout=None):
"""Stop the capture worker and the view pool, and forget them.
"""Stop the capture worker and the image pool, and forget them.
Bounded by ``timeout`` (see AutoThumbnailer.close). ``final`` (the
package teardown) also keeps a late monitor event from starting a
@@ -662,7 +674,7 @@ class AutoThumbnailMixin:
if final:
self._auto_thumbnails_shut = True
thumbnailer, self._auto_thumbnailer = self._auto_thumbnailer, None
pool, self._preview_pool_obj = self._preview_pool_obj, None
pool, self._image_pool_obj = self._image_pool_obj, None
self._preview_gate_obj = None
closed = True
if thumbnailer is not None:
@@ -706,16 +718,67 @@ class AutoThumbnailMixin:
gate = self._preview_gate_obj = _PreviewGate(ON_DEMAND_LIMIT, self.logger)
return gate
def _preview_pool(self):
with AutoThumbnailMixin._auto_thumbnailer_lock:
if self._preview_pool_obj is None:
self._preview_pool_obj = ThreadPoolExecutor(
ON_DEMAND_LIMIT, thread_name_prefix=POOL_PREFIX
)
return self._preview_pool_obj
def _image_pool(self):
"""The interactive image pool (#523): IMAGE_POOL_SIZE threads for view
previews (behind their gate), card facts and inspector reads, so
image buffers never spread over the default executor's threads.
def _preview_pool_threads(self):
pool = self._preview_pool_obj
None after the package teardown (``close_auto_thumbnails(final=True)``):
a late request then runs on the default executor (``run_in_executor``
with None) and starts no thread the teardown no longer stops.
"""
with AutoThumbnailMixin._auto_thumbnailer_lock:
if self._image_pool_obj is None and not self._auto_thumbnails_shut:
self._image_pool_obj = ThreadPoolExecutor(
IMAGE_POOL_SIZE, thread_name_prefix=POOL_PREFIX
)
return self._image_pool_obj
async def _run_image(self, func, *args, **kwargs):
"""``_run_in_executor`` for interactive image work: the image pool.
After the teardown, or with the pool shut down in between (a test
reset), the call runs on the default executor as before.
"""
call = functools.partial(func, *args, **kwargs)
loop = asyncio.get_running_loop()
try:
future = loop.run_in_executor(self._image_pool(), call)
except RuntimeError: # submit refused: the pool was shut down
future = loop.run_in_executor(None, call)
return await future
def _map_image(self, func, items):
"""``func`` over *items* on the image pool, in order (blocking).
Called from an image pool thread, after the teardown, or with the
pool shut down in between (its queued items are cancelled), the
items it did not run run on the calling thread.
"""
items = list(items)
pool = None if not items or _on_image_thread() else self._image_pool()
if pool is None:
return [func(item) for item in items]
futures = []
try:
for item in items:
futures.append(pool.submit(func, item))
except RuntimeError:
pass # shut down in between: the rest run here
results = []
for index, item in enumerate(items):
future = futures[index] if index < len(futures) else None
if future is None or future.cancelled():
results.append(func(item))
continue
try:
results.append(future.result())
except CancelledError: # cancelled while this caller waited
results.append(func(item))
return results
def _image_pool_threads(self):
pool = self._image_pool_obj
return [] if pool is None else list(getattr(pool, "_threads", ()))
def _video_placeholder_bytes(self):
@@ -730,7 +793,7 @@ class AutoThumbnailMixin:
return await self._preview_gate().run(
flight_key(Path(dst).parent, (Path(dst).name,)),
functools.partial(self._ensure_view, src, dst, is_video),
self._preview_pool(),
self._image_pool(),
)
def _ensure_view(self, src, dst, is_video):
@@ -774,13 +837,13 @@ class AutoThumbnailMixin:
The flight answers only the outcome (a history flight may be the one
joined); the file it previews is derived here: the preview path is
the request's own, and a fallback looks its source up again on the
view pool (bounded by it, never the default executor).
image pool (bounded by it, never the default executor).
"""
if _has_traversal(filepath) or not _is_preview_request(filepath):
return None
parts = tuple(filepath.split("/"))
gate = self._preview_gate()
pool = self._preview_pool()
pool = self._image_pool()
for root in output_dirs:
root = Path(root)
outcome = await gate.run(
+1
View File
@@ -153,6 +153,7 @@ class ThumbnailConvertMixin:
job.log_line("Conversion failed · internal error")
state = "failed"
job.finish(state)
await self._release_job_memory()
async def _convert_job_body(self, job, options):
output_dirs = await self._run_in_executor(self._get_all_output_dirs)
+1
View File
@@ -226,6 +226,7 @@ class ThumbnailJobsMixin:
)
state = "failed"
job.finish(state)
await self._release_job_memory()
async def _thumbnail_job_body(self, job, options):
"""Plan, then generate in worker batches; returns the final state.
+16 -1
View File
@@ -59,6 +59,9 @@ REWRITE_FAILED = (
)
RECLAIM_FAILED = "Reclaim space failed; the database is unchanged."
ROW_KEPT = "could not be rewritten; it keeps its inline text and reads as before"
# A protected value's row (#586), in the review's "not text; kept as is".
VALUE_KEPT = "stored value not text; kept as is"
KEPT_COLUMNS = {"workflow_data": "workflow", "prompt_metadata": "prompt graph"}
NOTHING_LEFT = "Every image record already stores its workflow once"
@@ -177,7 +180,19 @@ class WorkflowStorageRoutesMixin:
protected += batch["protected"]
errors += len(batch["errors"])
job.record_counted("changed", batch["rows"])
job.record_counted("protected", batch["protected"])
job.record_counted(
"protected",
batch["protected"],
[
{
"path": "",
"detail": f"Image record {image_id}: "
f"{KEPT_COLUMNS.get(column, column)}, {VALUE_KEPT}",
}
for image_id, column in batch["kept"]
],
complete=True,
)
for image_id in batch["errors"]:
job.record(
"errors",
+41
View File
@@ -320,3 +320,44 @@ test('copy toast and fallback dialog',async({page})=>{
await expect(fallback).toContainText('Automatic clipboard access requires a secure browser context.');
await page.keyboard.press('Escape');await expect(fallback).toBeHidden();await expect(copy).toBeFocused();
});
// #542 D191: "All N eligible nodes" previews every node, sends one after another and names each refusal.
test('All eligible nodes previews each node, sends in node id order and names a refusal',async({page,context},testInfo)=>{
const host=await context.newPage();await host.goto('/prompt_manager/workspace#/overview');
await host.evaluate(async()=>{
const {createBridgeHost}=await import('/prompt_manager/js/shared/toolkit/prompt-bridge.js');
const targets=['12','3','7'].map(id=>({id,label:id==='7'?'Edited in ComfyUI':'Prompt Manager (+/-)',type:'PromptManagerPair',roles:['positive','negative']}));
window.hostValues=Object.fromEntries(targets.map(target=>[target.id,{positive:`Existing ${target.id}`,negative:'Existing negative'}]));window.applied=[];
window.fixtureHost=createBridgeHost({origin:location.origin,namespace:'/',label:'Test ComfyUI',adapter:{
listTargets:()=>targets,
preview:(id,roles)=>{const target=targets.find(item=>item.id===id);return{preview:{target,fields:Object.entries(roles).map(([role,after])=>({role,before:window.hostValues[id][role],after}))},snapshot:{id,roles}};},
apply:({id,roles})=>{if(id==='7')throw new Error('Workflow or prompt changed. Preview again.');Object.assign(window.hostValues[id],roles);window.applied.push(id);return{targetId:id,values:roles};},
}});
});
try {
await page.goto(workspace);await select(page,'0:shared/same.png');await details(page).getByRole('button',{name:/^Send prompt pair to ComfyUI: /}).click();
const sender=dialog(page);const nodes=sender.getByRole('combobox',{name:'Destination node',exact:true});
await sender.getByRole('combobox',{name:'ComfyUI window',exact:true}).selectOption({label:'Test ComfyUI'});
await expect(nodes.locator('option')).toHaveText(['Choose a destination node','#12 Prompt Manager (+/-)','#3 Prompt Manager (+/-)','#7 Edited in ComfyUI','All 3 eligible nodes']);
await expect(nodes).toHaveValue('');
await nodes.selectOption({label:'All 3 eligible nodes'});
await expect(sender.getByRole('status')).toHaveText('Sending to all 3 eligible nodes in Test ComfyUI. Apply to send it, or preview the changes first.');
await page.screenshot({path:testInfo.outputPath('send-all-option.png')});
await sender.getByRole('button',{name:'Preview changes',exact:true}).click();
const reviews=sender.locator('section.prompt-sender-node');
await expect(reviews.locator('> h4')).toHaveText(['#3 Prompt Manager (+/-)','#7 Edited in ComfyUI','#12 Prompt Manager (+/-)']);
await expect(sender.getByRole('region',{name:'#12 Prompt Manager (+/-)',exact:true})).toContainText('Existing 12');
expect(await host.evaluate(()=>window.applied)).toEqual([]);
await sender.getByRole('button',{name:'Apply to ComfyUI',exact:true}).click();
await expect(sender.getByRole('alert')).toHaveText('Sent to 2 of 3 nodes in Test ComfyUI. #7 Edited in ComfyUI: Workflow or prompt changed. Preview again.');
expect(await host.evaluate(()=>window.applied)).toEqual(['3','12']);
expect(await host.evaluate(()=>window.hostValues['7'].positive)).toBe('Existing 7');
await page.screenshot({path:testInfo.outputPath('send-all-result.png')});
await page.setViewportSize({width:390,height:844});
await sender.getByRole('button',{name:'Preview changes',exact:true}).click();await expect(reviews).toHaveCount(3);
await reviews.last().scrollIntoViewIfNeeded();
expect(await page.evaluate(()=>document.querySelector('dialog.prompt-sender').scrollWidth<=document.querySelector('dialog.prompt-sender').clientWidth+1)).toBe(true);
await page.screenshot({path:testInfo.outputPath('send-all-mobile.png')});
const apply=sender.getByRole('button',{name:'Apply to ComfyUI',exact:true});await apply.focus();await expect(apply).toBeFocused();
} finally {await host.evaluate(()=>window.fixtureHost.dispose());await host.close();}
});
@@ -64,11 +64,31 @@ function measureMaintenance(mainSelector) {
};
}
/** D102: the Metadata cache row says why its review is unavailable and its button is aria-disabled (D93's dimmed style). */
function mapMaintenanceConcept(reference) {
/**
* D192 (#550, beyond the concept): Database adds "Pair category and tags" after "Unused prompt components", shaped like
* the concept's plain Review plan row (Optimize database's).
*/
function mapPairRow(reference) {
return {
...reference,
surfaces: reference.surfaces.map(surface => ({ ...surface, rows: surface.rows.map(row => row.title !== 'Metadata cache' ? row : {
surfaces: reference.surfaces.map(surface => {
const plain = surface.rows.find(row => row.title === 'Optimize database');
const at = surface.rows.findIndex(row => row.title === 'Unused prompt components');
if (!plain || at < 0) return surface;
return { ...surface, tree: [...surface.tree, 'div.operation'], rows: [...surface.rows.slice(0, at + 1),
{ ...plain, title: 'Pair category and tags', description: 'Fill an empty category or tags on prompts older ' +
'Pair node runs saved, from their saved pairs. Only empty fields of never-edited prompts.' },
...surface.rows.slice(at + 1)] };
}),
};
}
/** D102: the Metadata cache row says why its review is unavailable and its button is aria-disabled (D93's dimmed style). */
function mapMaintenanceConcept(reference) {
const mapped = mapPairRow(reference);
return {
...mapped,
surfaces: mapped.surfaces.map(surface => ({ ...surface, rows: surface.rows.map(row => row.title !== 'Metadata cache' ? row : {
...row, description: `${row.description} Nothing is cached in this browser.`,
actions: row.actions.map(action => ({ ...action, opacity: '0.55', cursor: 'not-allowed' })) }) })),
};
@@ -209,8 +229,8 @@ test('maintenance comparison detects mutations', async ({ page, request }) => {
const expected = mapMaintenanceConcept(concept);
const production = await productionMaintenance(page, 'dark');
expect(diffInventory(expected, production)).toEqual([]);
// Unmapped, the section differs only on the D102 row.
expect(diffInventory(concept, production).map(item => item.path).sort()).toEqual([
// Mapped only for the D192 row, the section differs only on the D102 row.
expect(diffInventory(mapPairRow(concept), production).map(item => item.path).sort()).toEqual([
'surfaces.1.rows.3.actions.0.cursor', 'surfaces.1.rows.3.actions.0.opacity', 'surfaces.1.rows.3.description']);
const measure = () => page.evaluate(measureMaintenance, '#workspace-main');
const paths = async () => diffInventory(expected, await measure()).map(item => item.path);
+11 -2
View File
@@ -92,6 +92,8 @@ export function cardActionsInventory(card) {
};
}
// #431: the fixture's saved "Describe the image" entry, the default of both engines (workspace_tagging_fixture.py).
export const FIXTURE_INSTRUCTIONS_ID='d35c71be0a1e';
// D84 (plan D61): the legacy engines' values, labels and sizes, GGUF JoyCaption in the concept's Ollama slot.
const MODEL_LABELS=Object.freeze({'wd14-swinv2':['wd14-swinv2','WD14 SwinV2 · ~400 MB'],'wd14-vit':['wd14-vit','WD14 ViT · ~400 MB'],
ollama:['gguf','GGUF JoyCaption (Recommended) · ~4.5 GB'],huggingface:['hf','HuggingFace JoyCaption · ~16 GB']});
@@ -106,17 +108,24 @@ const CONCEPT_LABELS=Object.freeze({'WD14 SwinV2 · fast classifier':'WD14 SwinV
* (`chip`: a concept `.chip` inventory, the heading's size chip) around the row's own text (12px) in the checked
* row's colours (accent border, `--soft-blue`);
* - D84 (plan D61): the legacy engine labels and values, and the server's default instructions (`instructions`);
* - D204 (#431): a "Saved instructions" select (`picker`: its value and options) before the instructions, on the
* concept's select field tokens;
* - D91 (plan D68): a Suggestions count field after the model's own fields, on the concept's number field tokens
* (`numberField`: the concept's General confidence field) with the configured count (`limit`);
* - D29: the prototype's "Sample suggestions from …" status reads "Suggestions from …".
* The opened state is compared unmapped: both show the concept's on-screen "Generate suggestions with the selected
* model before approving." (its renderTagModelOptions, a1062–1076, replaces the markup's sentence at once).
*/
export function mapConceptTagging(reference,{instructions,limit=10,chip,numberField}) {
export function mapConceptTagging(reference,{instructions,limit=10,chip,numberField,
picker={value:FIXTURE_INSTRUCTIONS_ID,options:[['builtin','Built-in'],[FIXTURE_INSTRUCTIONS_ID,`${instructions} · default`]]}}) {
const [select,...own]=reference.modelPanel.fields;
// D204 (#431): a "Saved instructions" select on the concept's select field, before the instructions.
const pickField={...select,label:'Saved instructions',control:{...select.control,id:'image-tag-instructions-pick',
value:picker.value,options:picker.options}};
const suggestionsField={...numberField,label:'Suggestions',control:{...numberField.control,id:'image-suggestion-limit',
value:String(limit),min:'1',max:'100',step:'1'}};
const mappedOwn=own.map(field=>field.control.id==='image-tag-instructions'?{...field,control:{...field.control,value:instructions}}:field);
const mappedOwn=own.flatMap(field=>field.control.id==='image-tag-instructions'
?[pickField,{...field,control:{...field.control,value:instructions}}]:[field]);
const rows=reference.suggestions.rows.map(row=>({...row,
control:row.control?.kind==='checkbox'&&row.control.checked?{kind:'remove',label:`Remove ${row.name}`}:row.control,
borderTopLeftRadius:chip.borderTopLeftRadius}));
@@ -284,6 +284,23 @@ test('a restore seen finishing elsewhere closes an open restore review and disca
expect((await serverJson(request, '/__test__/system/backups')).stages).toBe(0);
});
// #583: in WebKit, while a page.route is armed, a file part of a FormData body is read only when the request goes on;
// a garbage collection in between dropped the form once nothing referenced it, and the server got an empty file
// ("Uploaded file is empty"). The page keeps the form until its upload is answered. The route holds the upload
// through a collection; without the fix WebKit lost the file every time (0/10 in the pinned image). Chromium and
// Firefox keep the body anyway, so there it only checks that a held upload still opens the review.
test('a review upload survives a garbage collection while it is in flight', async ({ page, request }) => {
const bytes = await backupBytes(request);
await page.route(`**${PREVIEW_URL}`, async route => {
await page.requestGC();
await route.continue();
});
await openBackups(page);
expect((await chooseBackup(page, bytes)).status()).toBe(200);
await expect(dialog(page).locator('#modal-title')).toHaveText('Restore preview');
expect((await serverJson(request, '/__test__/system/backups')).stages).toBe(1);
});
test('invalid file shows the validator message; nothing is staged', async ({ page, request }) => {
const before = await serverJson(request, '/__test__/system/digest');
await openBackups(page);
@@ -64,7 +64,9 @@ test('the section shows the concept rows; Metadata cache follows Q5 (D102)', asy
item.querySelector('.row-actions > button').textContent, item.querySelector('.row-actions > button').className]),
]))).toEqual([
['Database', [['Integrity & relationship checks', 'Check', ''], ['Duplicate prompt text', 'Review plan', ''],
['Optimize database', 'Review plan', ''], ['Unused prompt components', 'Review plan', 'danger']]],
['Optimize database', 'Review plan', ''], ['Unused prompt components', 'Review plan', 'danger'],
// D192 (#550): beyond the concept.
['Pair category and tags', 'Review plan', '']]],
['Images & cache', [['Missing-file links', 'Review plan', 'danger'], ['Duplicate image files', 'Find duplicates', ''],
['Thumbnail cache', 'Review options', ''], ['Metadata cache', 'Review plan', '']]],
]);
@@ -177,6 +179,70 @@ test("plan dialog shows the server's scope, protected, backup policy and media l
expect(await serverJson(request, '/__test__/system/digest')).toEqual(before);
});
test('D192: Pair category and tags reviews old Pair prompts, keeps an edited one and fills the other (#550)',
async ({ page, request }) => {
await expect(await request.post('/__test__/system/pair-metadata')).toBeOK();
const answers = [];
page.on('response', response => { if (response.url().endsWith(PREVIEW_URL)) answers.push(response.json()); });
await openMaintenance(page);
const pairRow = row(page, 'Pair category and tags');
await expect(pairRow.locator('p')).toHaveText('Fill an empty category or tags on prompts older Pair node runs saved, ' +
'from their saved pairs. Only empty fields of never-edited prompts.');
await pairRow.getByRole('button', { name: 'Review plan', exact: true }).click();
const plan = dialog(page);
await expect(plan.locator('#modal-body > h3')).toHaveText('Pair category and tags');
const answer = await answers[0];
expect(answer.operation).toBe('pair_metadata');
expect(await gridOf(page)).toEqual([
['Scope', '1 prompt to fill: 1 category, 1 tag set'],
['Protected', '1 edited after it was saved'],
['Backup', `Create and verify before changes (${answer.backup.destination_label})`],
['Media on disk', 'No image file or saved pair is changed.'],
]);
await expect(plan.locator('details > summary')).toHaveText('Listed records (3)');
await plan.locator('details > summary').click();
// Protected rows come first, the Record row last.
const lines = (await plan.locator('details > pre').textContent()).split('\n');
expect(lines).toHaveLength(3);
expect(lines[0]).toBe('S550 harbor · Kept: edited after it was saved; its empty category or tags may be intentional');
expect(lines[1]).toMatch(/^S550 lantern · Fills category “night” and 2 tags: s550-glow, s550-warm \(pair \d+\)$/);
expect(lines[2]).toMatch(/^Record · The job's report lists every filled field\. Download it to keep it/);
await plan.getByRole('button', { name: 'Apply reviewed operation', exact: true }).click();
await expect(dialog(page)).toHaveCount(0);
await expect(page).toHaveURL(/#\/activity\/[0-9a-f]{32}$/);
const job = await ended(request, page.url().split('/').pop());
expect([job.title, job.state, job.counters.changed, job.counters.protected])
.toEqual(['Pair category and tags', 'finished', 1, 1]);
const main = page.locator('#workspace-main');
await expect(main.locator('h1')).toHaveText('Pair category and tags');
await expect(main.getByRole('button', { name: 'Download report', exact: true })).toBeVisible();
// #550 live test: the Changed group lists the path-less record rows as plain text, never an open action.
const results = main.getByRole('region', { name: 'Results', exact: true });
await results.locator('.outcome-list > button').filter({ hasText: 'Changed' }).click();
const records = results.locator('.outcome-files > .file-row');
await expect(records).toHaveCount(1);
await expect(records.first()).toHaveText(/^Prompt \d+: category “night”; tags s550-glow, s550-warm$/);
expect(await records.first().evaluate(node => node.localName)).toBe('div');
await expect(results.locator('.outcome-files button')).toHaveCount(0);
await expect(results.getByRole('searchbox', { name: 'Filter results' })).toHaveAttribute('placeholder', 'Filter records…');
// Protected lists each kept prompt with the review's reason (#550 live re-check).
await results.locator('.outcome-list > button').filter({ hasText: 'Protected' }).click();
await expect(records).toHaveCount(1);
await expect(records.first()).toHaveText(
/^Prompt \d+: kept, edited after it was saved; its empty category or tags may be intentional$/);
// An empty group says records, not files.
await results.locator('.outcome-list > button').filter({ hasText: 'Errors' }).click();
await expect(results.locator('.outcome-files > p.empty-state')).toHaveText('No records in this group.');
const library = await serverJson(request, '/prompt_manager/library');
const lantern = library.items.find(item => item.text === 'S550 lantern');
const harbor = library.items.find(item => item.text === 'S550 harbor');
const filled = (await serverJson(request, `/prompt_manager/library/${lantern.id}?archive=all`)).positive;
expect([filled.category, [...filled.tags].sort()]).toEqual(['night', ['s550-glow', 's550-warm']]);
expect((await serverJson(request, `/prompt_manager/library/${harbor.id}?archive=all`)).positive.category).toBeFalsy();
await page.goto(`${workspace}#/library`);
await expect(page.locator(`article[data-prompt-id="${lantern.id}"] .card-title > .chip`)).toHaveText('night');
});
test('a plan with nothing to do shows only Close, never Apply', async ({ page, request }) => {
const applies = [];
page.on('request', item => { if (item.url().endsWith(APPLY_URL)) applies.push(item.url()); });
@@ -341,3 +341,21 @@ test('a running scan makes the start report busy', async ({ page, request }) =>
await request.post(`/__test__/jobs/${scan.job.id}/finish`, { data: { state: 'finished' } });
}
});
// D204 (#431): Auto-tag selected has the same picker; the job sends the text the dialog shows.
test('Auto-tag selected picks saved instructions and sends the text shown (D204)', async ({ page }) => {
await select(page, FIRST);
await barButton(page, 'Auto-tag selected').click();
await expect(title(page)).toHaveText('Auto-tag selected images');
const body = modal(page).locator('#modal-body');
await body.getByRole('combobox', { name: 'Tagging model', exact: true }).selectOption('gguf');
const pick = body.getByRole('combobox', { name: 'Saved instructions', exact: true });
await expect(pick).toHaveAttribute('id', 'bulk-tag-instructions-pick');
await expect(pick.locator('option')).toHaveText(['Built-in', 'Describe the image · default']);
const text = body.getByRole('textbox', { name: 'Tagging instructions', exact: true });
await text.fill('Bulk one-off.');
await expect(pick.locator('option:checked')).toHaveText('Edited for this run');
const started = jobStart(page);
await actionButtons(page).filter({ hasText: 'Start auto-tagging' }).click();
expect((await started).postDataJSON().options.instructions).toBe('Bulk one-off.');
});
+47 -1
View File
@@ -4,7 +4,7 @@
import { startCoverage, saveCoverage } from '../../tools/vendor/browser-coverage.mjs';
import { test, expect } from './fixtures.mjs';
import { saveImageTags } from './concept-inventory.mjs';
import { taggingInventory } from './tagging-canonical-comparison.mjs';
import { FIXTURE_INSTRUCTIONS_ID, taggingInventory } from './tagging-canonical-comparison.mjs';
const images = '/prompt_manager/workspace#/images';
const MODULE = '/prompt_manager/js/workspace/image-tagging.mjs';
@@ -861,3 +861,49 @@ test('suggestions omit skipped tags and say how many were hidden', async ({ page
await expect(chips(page).locator(':scope > span:first-child')).toHaveText(['sandstone', 'arch', 'sky', 'dune']);
await expect(dialog(page).locator('#tag-suggestions > p.small')).toHaveText(`${CHIP_HINT} 2 skipped tags were not suggested.`);
});
// D204 (#431): the Saved instructions select above Tagging instructions: the engine default, a pick, a one-off edit
// ("Edited for this run" and the muted note), an engine switch resets, WD14 hides both; Generate sends the text shown.
test('the saved instructions picker fills, edits one-off and resets per engine (D204)', async ({ page, request }) => {
const settings = await (await request.get('/prompt_manager/settings')).json();
const builtin = settings.settings.autotag_builtin_instructions;
await openTag(page);
const options = dialog(page).locator('#image-tag-model-options');
const pick = options.getByRole('combobox', { name: 'Saved instructions', exact: true });
const text = options.getByRole('textbox', { name: 'Tagging instructions', exact: true });
const note = options.locator('.instructions-edited');
await expect(pick).toHaveCount(0);
await model(page).selectOption('gguf');
await expect(pick.locator('option')).toHaveText(['Built-in', 'Describe the image · default']);
await expect(pick).toHaveAttribute('id', 'image-tag-instructions-pick');
await expect(text).toHaveValue('Describe the image');
await expect(note).toBeHidden();
// A keyboard pick (the select focused) keeps focus on the select the fields are re-rendered with.
await pick.focus();
await pick.selectOption('builtin');
await expect(text).toHaveValue(builtin);
await expect(pick).toBeFocused();
await text.fill('Only the sky.');
await expect(text).toHaveAttribute('aria-describedby', 'image-tag-instructions-pick-note');
await expect(pick.locator('option:checked')).toHaveText('Edited for this run');
await expect(pick.locator('option[value="edited"]')).toBeDisabled();
await expect(note).toHaveText('Edits apply to this run only. Manage saved instructions in System › Auto-tagging.');
await expect(text).toBeFocused();
// Generate sends the edited text; the saved entries are unchanged.
const sent = page.waitForRequest(item => item.url().endsWith('/image-annotations/suggest'));
await generate(page).click();
expect((await sent).postDataJSON().options.instructions).toBe('Only the sky.');
// Choosing an entry replaces the edit; switching engines starts from the new engine's default.
await pick.selectOption('builtin');
await expect(text).toHaveValue(builtin);
await expect(note).toBeHidden();
await text.fill('Edited again.');
await model(page).selectOption('hf');
await expect(text).toHaveValue('Describe the image');
await expect(pick).toHaveValue(FIXTURE_INSTRUCTIONS_ID);
await model(page).selectOption('wd14-vit');
await expect(pick).toHaveCount(0);
await expect(text).toHaveCount(0);
expect((await (await request.get('/prompt_manager/settings')).json()).settings.autotag_instructions)
.toEqual(settings.settings.autotag_instructions);
});
+24
View File
@@ -41,6 +41,11 @@ read. These per-server hooks only change what a System test needs on top:
- ``POST /__test__/system/duplicate-prompts`` (S7) adds two prompts whose
text differs only in case and spacing (a duplicate prompt text group);
``{"again": true}`` adds a third, so a reviewed plan no longer matches;
- ``POST /__test__/system/pair-metadata`` (#550) saves two saved pairs the
way the Pair node did before #546 (the pair holds the category and tags,
its new positive prompt none): ``S550 lantern`` (category ``night``, tags
``s550-glow``, ``s550-warm``) fills; ``S550 harbor`` was edited after it was
saved, so the plan keeps it;
- ``POST /__test__/system/duplicate-files`` (S7) writes one PNG's bytes three
times: ``output/s7-dupes/same.png``, ``input/s7-dupes/same.png`` (the same
relative path in the other source) and ``output/s7-dupes/copy.png``;
@@ -344,6 +349,25 @@ def register_system_commands(app, routes, api, root: Path):
conn.commit()
return web.json_response({"success": True})
@routes.post("/__test__/system/pair-metadata")
async def pair_metadata(request):
def seed():
api.db.save_prompt_pair(
"S550 lantern",
"S550 blurry",
category="night",
tags=["s550-glow", "s550-warm"],
)
api.db.save_prompt_pair("S550 harbor", category="night")
with api.db.model.get_connection() as conn:
conn.execute(
"UPDATE prompts SET updated_at='2099-01-01T00:00:00+00:00' "
"WHERE text='S550 harbor'"
)
await api._run_in_executor(seed)
return web.json_response({"success": True})
@routes.post("/__test__/system/duplicate-files")
async def duplicate_files(request):
body = await request.json() if request.can_read_body else {}
+31 -2
View File
@@ -131,11 +131,40 @@ test('native image inspector previews and sends a pair into real visible widgets
const twinId=await page.evaluate(async()=>{const {app}=await import('/scripts/app.js');const node=LiteGraph.createNode('PromptManagerPair');app.graph.add(node);return node.id;});
await find0.click();
const targets=sender.getByRole('combobox',{name:'Destination node',exact:true});
await expect(targets.locator('option')).toHaveCount(3);
await expect(targets.locator('option')).toHaveCount(4);
await expect(targets).toHaveValue('');
await expect(apply).toBeDisabled();
await expect(targets.locator('option').nth(2)).toHaveText(new RegExp(`^#${twinId} `));
await page.evaluate(async twinId=>{const {app}=await import('/scripts/app.js');app.graph.remove(app.graph.getNodeById(twinId));},twinId);
// #542 D191: "All 2 eligible nodes" writes both real nodes, one after another; a linked third node is not offered.
const linkedId=await page.evaluate(async()=>{
const {app}=await import('/scripts/app.js');const {convertToInput}=await import('/extensions/core/widgetInputs.js');
const node=LiteGraph.createNode('PromptManagerPair');app.graph.add(node);
const widget=node.widgets.find(widget=>widget.name==='text');
if(!node.inputs.some(input=>input.name==='text'))convertToInput(node,widget);
const source=LiteGraph.createNode('PrimitiveString');app.graph.add(source);
source.connect(0,node,node.inputs.findIndex(input=>input.name==='text'));
for(const each of app.graph._nodes.filter(item=>item.type==='PromptManagerPair')){const text=each.widgets.find(widget=>widget.name==='text');if(text)text.value=`Before all ${each.id}`;}
return node.id;
});
await find0.click();
await expect(targets.locator('option')).toHaveText(['Choose a destination node',new RegExp(`^#${id} `),new RegExp(`^#${twinId} `),'All 2 eligible nodes']);
await expect(sender.getByRole('status')).toContainText('1 other Prompt Manager node cannot receive this prompt.');
await targets.selectOption({label:'All 2 eligible nodes'});
await sender.getByRole('button',{name:'Preview changes',exact:true}).click();
await expect(sender.locator('section.prompt-sender-node > h4')).toHaveText([new RegExp(`^#${id} `),new RegExp(`^#${twinId} `)]);
await apply.click();
// The tab title gains ComfyUI's unsaved mark once the workflow changed, so the window is matched loosely.
await expect(sender.getByRole('status')).toHaveText(/^Sent to 2 of 2 nodes in .*ComfyUI\.$/);
await expect(apply).toBeDisabled();
const all=await page.evaluate(async ids=>{
const {app}=await import('/scripts/app.js');const output=(await app.graphToPrompt()).output;
return ids.map(nodeId=>{const node=app.graph.getNodeById(nodeId);const serialized=node.serialize();const index=node.widgets.findIndex(widget=>widget.name==='negative_text');
return{prompt:output[String(nodeId)]?.inputs.negative_text,serialized:serialized.widgets_values[index],text:node.widgets.find(widget=>widget.name==='text')?.value};});
},[id,twinId,linkedId]);
expect(all.slice(0,2)).toEqual([{prompt:negative,serialized:negative,text:positive},{prompt:negative,serialized:negative,text:positive}]);
expect(all[2].text).toBe(`Before all ${linkedId}`);
expect(await page.evaluate(()=>window.senderEffects.queue-window.senderSendBaseline.queue)).toBe(0);
await page.evaluate(async ids=>{const {app}=await import('/scripts/app.js');for(const nodeId of ids)app.graph.remove(app.graph.getNodeById(nodeId));},[twinId,linkedId]);
// #535: the status line says why a window cannot receive.
const find=sender.getByRole('button',{name:'Find ComfyUI windows',exact:true});
const preview=sender.getByRole('button',{name:'Preview changes',exact:true});
+138
View File
@@ -225,3 +225,141 @@ test('acknowledgement cannot claim an extra role beyond approved changes',async(
const {createSendSession}=await source;const s=setup(createSendSession,{apply:async()=>({targetId:'one',values:{positive:'new',negative:'also changed'}})});
await s.session.discover();s.session.selectHost('host');s.session.selectTarget('one');await s.session.preview();await s.session.apply();assert.equal(s.session.state.phase,'error');assert.match(s.session.state.message,/unexpected acknowledgement/);s.session.dispose();
});
// #542: "All N eligible nodes" sends to every eligible node of one window, previewing every node first (Q542-2 C).
function allHost({nodes=['10','9','2'],revision,preview,apply,discover}={}) {
const calls=[];const sent=new Set();
const targets=nodes.map(id=>pm(id,['positive'],revision?{revision}:{}));
const client={
discover:discover ?? (async()=>[window1(targets)]),
preview:async(host,id,roles)=>{calls.push(['preview',id]);if(preview)return preview(id,calls,sent);return{ticket:`t-${id}-${calls.length}`,revision,target:{id,label:`Node ${id}`},fields:[{role:'positive',before:sent.has(id)?'new':'old',after:roles.positive}]};},
apply:async(host,ticket)=>{const id=ticket.split('-')[1];calls.push(['apply',id,ticket]);if(apply)return apply(id,calls);sent.add(id);return{targetId:id,values:{positive:'new'}};},
dispose:()=>calls.push(['dispose']),
};
return {client,calls,sent};
}
async function allSession(createSendSession,options) {
const host=allHost(options);const session=createSendSession({client:host.client,roles:{positive:'new'},onChange:()=>{}});
await session.discover();return {...host,session};
}
const order=calls=>calls.filter(call=>call[0]!=='dispose').map(call=>`${call[0]} ${call[1]}`);
test('All is offered only with two or more eligible nodes and is never selected for you',async()=>{
const {createSendSession,ALL_NODES}=await source;
const one=await allSession(createSendSession,{nodes:['1']});
one.session.selectTarget(ALL_NODES);assert.equal(one.session.state.targetId,'');assert.equal(one.calls.length,0);one.session.dispose();
const two=await allSession(createSendSession,{nodes:['1','2']});
assert.equal(two.session.state.targetId,'');
two.session.selectTarget(ALL_NODES);assert.equal(two.session.state.targetId,ALL_NODES);
assert.match(two.session.state.message,/all 2 eligible nodes in Flow - ComfyUI/);
await two.session.discover();assert.equal(two.session.state.targetId,'');two.session.dispose();
});
test('Apply to All previews every node first, then sends one after another in node id order',async()=>{
const {createSendSession,ALL_NODES}=await source;const s=await allSession(createSendSession);
s.session.selectTarget(ALL_NODES);await s.session.apply();
assert.deepEqual(order(s.calls),['preview 2','preview 9','preview 10','apply 2','preview 9','apply 9','preview 10','apply 10']);
assert.equal(s.session.state.phase,'success');assert.equal(s.session.state.message,'Sent to 3 of 3 nodes in Flow - ComfyUI.');
await s.session.apply();assert.equal(s.calls.filter(call=>call[0]==='apply').length,3);s.session.dispose();
});
test('a double click on Apply to All sends to each node once',async()=>{
const {createSendSession,ALL_NODES}=await source;const s=await allSession(createSendSession,{nodes:['1','2']});
s.session.selectTarget(ALL_NODES);await Promise.all([s.session.apply(),s.session.apply()]);
assert.deepEqual(s.calls.filter(call=>call[0]==='apply').map(call=>call[1]),['1','2']);s.session.dispose();
});
test('a node refusing its preview stops All before anything is written and names the node',async()=>{
const {createSendSession,ALL_NODES}=await source;
const s=await allSession(createSendSession,{preview:id=>{if(id==='9')throw Error('positive is connected to an upstream input.');return{ticket:`t-${id}`,target:{id},fields:[{role:'positive',before:'old',after:'new'}]};}});
s.session.selectTarget(ALL_NODES);await s.session.apply();
assert.equal(s.calls.filter(call=>call[0]==='apply').length,0);assert.equal(s.session.state.phase,'error');
assert.match(s.session.state.message,/^#9 Node 9: positive is connected to an upstream input\. Nothing was sent\./);
assert.equal(s.session.state.outcomeUnknown,false);s.session.dispose();
});
test('a refusal while sending to All is reported by node and the other nodes are still sent',async()=>{
const {createSendSession,ALL_NODES}=await source;
const s=await allSession(createSendSession,{apply:id=>{if(id==='9')throw Error('Workflow or prompt changed. Preview again.');return{targetId:id,values:{positive:'new'}};}});
s.session.selectTarget(ALL_NODES);await s.session.apply();
assert.deepEqual(s.calls.filter(call=>call[0]==='apply').map(call=>call[1]),['2','9','10']);
assert.equal(s.session.state.phase,'error');assert.equal(s.session.state.outcomeUnknown,false);
assert.equal(s.session.state.message,'Sent to 2 of 3 nodes in Flow - ComfyUI. #9 Node 9: Workflow or prompt changed. Preview again.');s.session.dispose();
});
test('a lost reply on one node of All leaves the later nodes unsent and that outcome unknown',async()=>{
const {createSendSession,ALL_NODES}=await source;
const s=await allSession(createSendSession,{apply:id=>{if(id==='9')throw Error('Bridge request timed out');return{targetId:id,values:{positive:'new'}};}});
s.session.selectTarget(ALL_NODES);await s.session.apply();
assert.deepEqual(s.calls.filter(call=>call[0]==='apply').map(call=>call[1]),['2','9']);
assert.equal(s.session.state.outcomeUnknown,true);
assert.match(s.session.state.message,/^Sent to 1 of 3 nodes in Flow - ComfyUI\. No confirmation received from #9 Node 9; the update may have applied\. Not sent: #10 Node 10\. Check ComfyUI, then preview again\. Bridge request timed out$/);
await s.session.apply();assert.equal(s.calls.filter(call=>call[0]==='apply').length,2);s.session.dispose();
});
test('closing the dialog during All stops after the current node',async()=>{
const {createSendSession,ALL_NODES}=await source;let release;
const s=await allSession(createSendSession,{apply:id=>id==='2'?new Promise(done=>release=()=>done({targetId:id,values:{positive:'new'}})):{targetId:id,values:{positive:'new'}}});
s.session.selectTarget(ALL_NODES);const pending=s.session.apply();
for(let i=0;i<50 && !release;i++)await tick();assert.ok(release,'the first node was sent');
s.session.dispose();release();await pending;
assert.deepEqual(s.calls.filter(call=>call[0]==='apply').map(call=>call[1]),['2']);
});
test('All accepts the revision our own write moved, but not a workflow changed before the run',async()=>{
const {createSendSession,ALL_NODES}=await source;
// The host moves the graph version on every write: later nodes preview with the new revision.
let version=4;
const bumped=await allSession(createSendSession,{revision:'1:4',preview:id=>({ticket:`t-${id}`,revision:`1:${version}`,target:{id},fields:[{role:'positive',before:'old',after:'new'}]}),apply:id=>{version++;return{targetId:id,values:{positive:'new'}};}});
bumped.session.selectTarget(ALL_NODES);await bumped.session.apply();
assert.equal(bumped.session.state.message,'Sent to 3 of 3 nodes in Flow - ComfyUI.');bumped.session.dispose();
const moved=await allSession(createSendSession,{revision:'1:4',preview:id=>({ticket:`t-${id}`,revision:'1:5',target:{id},fields:[{role:'positive',before:'old',after:'new'}]})});
moved.session.selectTarget(ALL_NODES);await moved.session.apply();
assert.match(moved.session.state.message,/workflow changed since this window was found/i);assert.equal(moved.calls.filter(call=>call[0]==='apply').length,0);moved.session.dispose();
});
test('a node whose text changed after the preview pass is refused, not overwritten',async()=>{
const {createSendSession,ALL_NODES}=await source;let edited=false;
const s=await allSession(createSendSession,{preview:id=>({ticket:`t-${id}`,target:{id},fields:[{role:'positive',before:id==='10'&&edited?'typed in ComfyUI':'old',after:'new'}]}),apply:id=>{edited=true;return{targetId:id,values:{positive:'new'}};}});
s.session.selectTarget(ALL_NODES);await s.session.apply();
assert.deepEqual(s.calls.filter(call=>call[0]==='apply').map(call=>call[1]),['2','9']);
assert.match(s.session.state.message,/^Sent to 2 of 3 nodes in Flow - ComfyUI\. #10 Node 10: Its prompt changed since the preview\./);s.session.dispose();
});
test('Preview with All shows every node in order, and Apply sends fresh tickets that match the review',async()=>{
const {createSendSession,ALL_NODES}=await source;const s=await allSession(createSendSession);
s.session.selectTarget(ALL_NODES);await s.session.preview();
assert.equal(s.session.state.phase,'ready');assert.equal(s.session.state.preview,null);
assert.deepEqual(s.session.state.previews.map(item=>item.label),['#2 Node 2','#9 Node 9','#10 Node 10']);
assert.match(s.session.state.message,/Sending to all 3 eligible nodes in Flow - ComfyUI\. Review the complete changes, then apply\./);
const first=s.session.state.previews[0].preview.ticket;
await s.session.apply();
// A held review may have outlived the host's 30 s tickets, so even the first node takes a fresh one.
assert.deepEqual(order(s.calls),['preview 2','preview 9','preview 10','preview 2','apply 2','preview 9','apply 9','preview 10','apply 10']);
assert.notEqual(s.calls.find(call=>call[0]==='apply')[2],first);assert.equal(s.session.state.phase,'success');
s.session.selectTarget('2');assert.deepEqual(s.session.state.previews,[]);s.session.dispose();
});
test('the dialog offers All for two or more nodes and reviews each node under its name',async()=>{
const {createPromptSender,ALL_NODES}=await source;const {createScope}=await import('../../web/js/shared/toolkit/core.js');
const dom=fakeDOM();const scope=createScope();
const {client}=allHost({nodes:['1','2']});
try {
createPromptSender({scope,clientFactory:()=>client}).open({positive:'new'});await tick();
const dialog=dom.body.children.at(-1);const target=dom.find(dialog,node=>node.id==='prompt-sender-target');
assert.deepEqual(target.children.map(option=>[option.value,option.textContent]),[['','Choose a destination node'],['1','#1 Node 1'],['2','#2 Node 2'],[ALL_NODES,'All 2 eligible nodes']]);
assert.equal(target.value,'');target.value=ALL_NODES;target.dispatchEvent(new Event('change'));
dom.find(dialog,node=>node.textContent==='Preview changes').dispatchEvent(new Event('click'));for(let i=0;i<5;i++)await tick();
const sections=[];(function walk(node){if(node.className==='prompt-sender-node')sections.push(node);node.children.forEach(walk);})(dialog);
assert.deepEqual(sections.map(section=>section.children[0].textContent),['#1 Node 1','#2 Node 2']);
dom.find(dialog,node=>node.textContent==='Apply to ComfyUI').dispatchEvent(new Event('click'));for(let i=0;i<10;i++)await tick();
assert.ok(dom.find(dialog,node=>node.textContent==='Sent to 2 of 2 nodes in Flow - ComfyUI.'));
} finally {scope.dispose();dom.restore();}
});
test('an expired review still sends every node, and a held review checks the discovered revision first',async()=>{
const {createSendSession,ALL_NODES}=await source;let reviewing=true;
// The host refuses every ticket taken during the review, as after its 30 s expiry.
const s=await allSession(createSendSession,{preview:(id,calls)=>({ticket:`t-${id}-${reviewing?'old':calls.length}`,target:{id},fields:[{role:'positive',before:'old',after:'new'}]}),apply:id=>({targetId:id,values:{positive:'new'}})});
const realApply=s.client.apply;s.client.apply=async(host,ticket,options)=>{if(ticket.endsWith('-old'))throw Error('Preview expired or unavailable; preview again');return realApply(host,ticket,options);};
s.session.selectTarget(ALL_NODES);await s.session.preview();reviewing=false;await s.session.apply();
assert.equal(s.session.state.message,'Sent to 3 of 3 nodes in Flow - ComfyUI.');s.session.dispose();
const moved=await allSession(createSendSession,{revision:'1:4',preview:(id,calls)=>({ticket:`t-${id}`,revision:calls.length>3?'1:5':'1:4',target:{id},fields:[{role:'positive',before:'old',after:'new'}]})});
moved.session.selectTarget(ALL_NODES);await moved.session.preview();assert.equal(moved.session.state.phase,'ready');await moved.session.apply();
assert.match(moved.session.state.message,/^Sent to 0 of 3 nodes in Flow - ComfyUI\. #2 Node 2: The workflow changed since this window was found/);
assert.equal(moved.calls.filter(call=>call[0]==='apply').length,0);moved.session.dispose();
});
test('a changed prepend or append after the review refuses that node',async()=>{
const {createSendSession,ALL_NODES}=await source;let edited=false;
const s=await allSession(createSendSession,{preview:id=>({ticket:`t-${id}`,target:{id},fields:[{role:'positive',before:'old',after:'new'}],composition:{prepend:id==='9'&&edited?'typed':'trigger',append:'',before:'trigger old',after:'trigger new'}}),apply:id=>{edited=true;return{targetId:id,values:{positive:'new'}};}});
s.session.selectTarget(ALL_NODES);await s.session.apply();
assert.deepEqual(s.calls.filter(call=>call[0]==='apply').map(call=>call[1]),['2','10']);
assert.match(s.session.state.message,/^Sent to 2 of 3 nodes in Flow - ComfyUI\. #9 Node 9: Its prompt changed since the preview\./);s.session.dispose();
});
+92 -3
View File
@@ -53,8 +53,12 @@ function observed(overrides = {}, capture = {}) {
...overrides,
};
}
// #431: the saved instructions; the fixture's entry is the default of both engines.
const SAVED_ENTRY = { id: '0123456789ab', name: 'Saved', text: DEFAULT_PROMPT };
const models = (overrides = {}) => ({
success: true, models: { 'wd14-swinv2': { downloaded: true } }, default_prompt: DEFAULT_PROMPT,
success: true, models: { 'wd14-swinv2': { downloaded: true } },
instructions: [{ id: 'builtin', name: 'Built-in', text: 'Built-in text.' }, SAVED_ENTRY],
default_instructions: { gguf: SAVED_ENTRY.id, hf: SAVED_ENTRY.id },
wd14_general_threshold: 0.35, wd14_character_threshold: 0.85, suggestion_limit: 10, ...overrides,
});
const SANDSTONE = [
@@ -244,10 +248,13 @@ test('dialog starts on the saved model unless a preset or the user chose one', a
});
test('an unreadable or malformed models read falls back to the documented defaults', async () => {
for (const route of [() => fail(500, null), () => response(200, models({ default_prompt: 4, suggestion_limit: 0, wd14_general_threshold: 2, wd14_character_threshold: 'x' }))]) {
for (const route of [() => fail(500, null), () => response(200, models({ instructions: 4, suggestion_limit: 0, wd14_general_threshold: 2, wd14_character_threshold: 'x' }))]) {
const { c } = await opened({ [`GET ${MODELS}`]: route });
assert.equal(c.state.phase, 'ready');
assert.deepEqual({ ...c.state.options }, { general_threshold: 0.35, character_threshold: 0.85, instructions: '', suggestion_limit: 10 });
// #431: the client's Built-in copy (D204's read failure).
const { BUILTIN_INSTRUCTIONS } = await load();
assert.deepEqual({ ...c.state.options }, { general_threshold: 0.35, character_threshold: 0.85, instructions: BUILTIN_INSTRUCTIONS,
suggestion_limit: 10 });
}
});
@@ -950,3 +957,85 @@ test('the chip hint says how many skipped tags were not suggested (#431)', async
assert.equal(c.state.chipHint, hint, String(skipped));
}
});
// D204 (#431): the Saved instructions picker of both dialogs.
const PHOTO_ENTRY = { id: '0123456789ac', name: 'Photo', text: 'Photo tags.' };
const picker = (overrides = {}) => ({ [`GET ${MODELS}`]: () => response(200, models({ model_type: 'gguf',
instructions: [{ id: 'builtin', name: 'Built-in', text: 'Built-in text.' }, SAVED_ENTRY, PHOTO_ENTRY],
default_instructions: { gguf: SAVED_ENTRY.id, hf: PHOTO_ENTRY.id }, ...overrides })) });
test('the picker starts at the engine default; a pick replaces the text and an edit is one-off (D204)', async () => {
const { c } = await opened(picker());
assert.deepEqual([c.state.instructionId, c.state.options.instructions, c.state.instructionsEdited], [SAVED_ENTRY.id, DEFAULT_PROMPT, false]);
const revision = c.state.optionsRevision;
c.pickInstructions(PHOTO_ENTRY.id);
assert.deepEqual([c.state.instructionId, c.state.options.instructions, c.state.instructionsEdited], [PHOTO_ENTRY.id, 'Photo tags.', false]);
assert.equal(c.state.optionsRevision, revision + 1, 'a pick re-renders the textarea');
c.setOption('instructions', 'Photo tags, please.');
assert.equal(c.state.instructionsEdited, true);
assert.equal(c.state.optionsRevision, revision + 1, 'typing never re-renders');
// Choosing an entry again replaces the edit without asking; an unknown id changes nothing.
c.pickInstructions(PHOTO_ENTRY.id);
assert.deepEqual([c.state.options.instructions, c.state.instructionsEdited], ['Photo tags.', false]);
c.pickInstructions('nope');
assert.equal(c.state.instructionId, PHOTO_ENTRY.id);
// Switching engines resets to the new engine's default and drops a one-off edit; WD14 is never "edited".
c.setOption('instructions', 'edited');
c.setModel('hf');
assert.deepEqual([c.state.instructionId, c.state.options.instructions, c.state.instructionsEdited], [PHOTO_ENTRY.id, 'Photo tags.', false]);
c.setModel('wd14-vit');
assert.equal(c.state.instructionsEdited, false);
c.setModel('gguf');
assert.deepEqual([c.state.instructionId, c.state.options.instructions], [SAVED_ENTRY.id, DEFAULT_PROMPT]);
c.dispose();
});
test('a preset shows the entry with its text, else "Edited for this run" (D204)', async () => {
const { c } = await opened(picker(), { preset: { model: 'hf', options: { instructions: DEFAULT_PROMPT } } });
assert.deepEqual([c.state.instructionId, c.state.instructionsEdited], [SAVED_ENTRY.id, false]);
const { c: other } = await opened(picker(), { preset: { model: 'hf', options: { instructions: 'Something else' } } });
assert.deepEqual([other.state.instructionId, other.state.options.instructions, other.state.instructionsEdited],
[PHOTO_ENTRY.id, 'Something else', true]);
for (const controller of [c, other]) controller.dispose();
});
test('a models read without instructions offers Built-in only with the client copy (D204)', async () => {
const { BUILTIN_INSTRUCTIONS, readChoices } = await load();
assert.equal(BUILTIN_INSTRUCTIONS.startsWith('Generate only comma-separated Danbooru tags'), true);
for (const value of [null, {}, { instructions: [] }, { instructions: [{ id: 'x', name: 'X', text: 'y' }] }]) {
assert.deepEqual(readChoices(value).entries.map(entry => entry.id), ['builtin'], JSON.stringify(value));
}
// A default naming no entry is Built-in.
assert.deepEqual({ ...readChoices({ instructions: [{ id: 'builtin', name: 'Built-in', text: 't' }], default_instructions: { gguf: 'gone' } }).defaults },
{ gguf: 'builtin', hf: 'builtin' });
const { c } = await opened(picker({ instructions: 'broken' }));
assert.deepEqual([c.state.instructionId, c.state.options.instructions], ['builtin', BUILTIN_INSTRUCTIONS]);
c.dispose();
});
test('the bulk options pick and reset like the Tag image dialog (D204)', async () => {
const { createTaggingOptionsController } = await load();
const options = createTaggingOptionsController({ api: harness(defaultRoutes(picker())).api });
await options.open();
assert.deepEqual([options.state.model, options.state.instructionId, options.state.options.instructions], ['gguf', SAVED_ENTRY.id, DEFAULT_PROMPT]);
options.pickInstructions(PHOTO_ENTRY.id);
assert.equal(options.request().options.instructions, 'Photo tags.');
options.setOption('instructions', 'One-off.');
assert.equal(options.state.instructionsEdited, true);
assert.equal(options.request().options.instructions, 'One-off.');
options.setModel('hf');
assert.deepEqual([options.state.instructionId, options.state.instructionsEdited], [PHOTO_ENTRY.id, false]);
options.pickInstructions('nope');
options.dispose();
options.pickInstructions(SAVED_ENTRY.id);
assert.equal(options.state.instructionId, PHOTO_ENTRY.id);
});
test('the client Built-in copy equals the server DEFAULT_TAG_TEMPLATE (#431)', async () => {
const { BUILTIN_INSTRUCTIONS } = await load();
const source = require('node:fs').readFileSync(require('node:path').join(__dirname, '../../py/tagging_defaults.py'), 'utf8');
const block = source.slice(source.indexOf('DEFAULT_TAG_TEMPLATE = ('), source.indexOf('\n)\n', source.indexOf('DEFAULT_TAG_TEMPLATE = (')));
const text = [...block.matchAll(/"((?:[^"\\]|\\.)*)"/g)].map(([, part]) => JSON.parse(`"${part}"`)).join('');
assert.ok(text.length > 100);
assert.equal(BUILTIN_INSTRUCTIONS, text);
});
@@ -180,7 +180,9 @@ test('the bulk options start from the server defaults and wait for them before S
assert.equal(options.state.canStart, false);
assert.equal(options.state.canGenerate, false);
const opening = options.open();
answer({ wd14_general_threshold: 0.5, wd14_character_threshold: 0.9, default_prompt: 'Say it', suggestion_limit: 7,
answer({ wd14_general_threshold: 0.5, wd14_character_threshold: 0.9, suggestion_limit: 7,
instructions: [{ id: 'builtin', name: 'Built-in', text: 'Built-in text.' }, { id: '0123456789ab', name: 'Say', text: 'Say it' }],
default_instructions: { gguf: '0123456789ab', hf: 'builtin' },
models: { 'wd14-swinv2': { downloaded: true }, 'wd14-vit': { downloaded: false }, gguf: { downloaded: true } } });
await opening;
assert.equal(options.state.canStart, true);
+18
View File
@@ -128,6 +128,24 @@ test('invalid file shows the validator message; a refusal names its job; a lost
assert.deepEqual(await uploadRestore(file, { ApiError, fetchUrl: async () => response(200, PREVIEW) }), PREVIEW);
});
// #583: WebKit reads a FormData's file part by reference, possibly after a garbage collection; the page holds the form
// until the upload is answered, then lets it go, whether the server staged it, refused it or the request failed.
test('a review upload keeps its form referenced until the upload is answered', async () => {
const { uploadRestore, uploadsInFlight } = await backups;
const file = new File(['bytes'], 'b.db');
for (const answer of [async () => response(200, PREVIEW), async () => response(400, { success: false, error: 'x' }),
async () => { throw new TypeError('offline'); }]) {
let held;
const upload = uploadRestore(file, { ApiError, fetchUrl: async (url, init) => {
held = uploadsInFlight.has(init.body) && init.body.get('database_file').size === file.size;
return answer();
} }).catch(() => {});
await upload;
assert.equal(held, true);
assert.equal(uploadsInFlight.size, 0);
}
});
test('Cancel discards the stage through the api client and never throws', async () => {
const { discardRestore, PREVIEW_URL } = await backups;
const calls = [];
+54 -2
View File
@@ -21,8 +21,23 @@ test('the section lists the concept rows in its order, labels and danger marks',
assert.deepEqual(MAINTENANCE_SURFACES.map(([heading]) => heading), surfaces.map(text => text.replace('&amp;', '&')));
const production = MAINTENANCE_SURFACES.flatMap(([, rows]) => rows.map(([title, description, label, , danger = false]) =>
[title, description, label, danger]));
assert.deepEqual(production, concept.map(([title, ...rest]) => [title.replace('&amp;', '&'), ...rest]));
assert.equal(production.length, 8);
// D192 (#550, beyond the concept): "Pair category and tags" follows "Unused prompt components" in Database.
const expected = concept.map(([title, ...rest]) => [title.replace('&amp;', '&'), ...rest]);
expected.splice(expected.findIndex(([title]) => title === 'Unused prompt components') + 1, 0, PAIR_ROW);
assert.deepEqual(production, expected);
assert.equal(production.length, 9);
});
const PAIR_ROW = Object.freeze(['Pair category and tags', 'Fill an empty category or tags on prompts older Pair node runs ' +
'saved, from their saved pairs. Only empty fields of never-edited prompts.', 'Review plan', false]);
test('D192: the Pair category and tags row reviews the pair_metadata plan under its title (#550)', async () => {
const { MAINTENANCE_SURFACES } = await maintenance;
const { OPERATION_TITLES } = await review;
const [, database] = MAINTENANCE_SURFACES[0];
const row = database.find(([title]) => title === PAIR_ROW[0]);
assert.deepEqual(row, [PAIR_ROW[0], PAIR_ROW[1], PAIR_ROW[2], 'pair_metadata']);
assert.equal(OPERATION_TITLES.pair_metadata, 'Pair category and tags');
});
const PLAN = Object.freeze({
@@ -210,3 +225,40 @@ test('a rotated-log cleanup the log overtook offers Review again (S12)', async (
assert.deepEqual(reviewAgainOf({ kind: 'maintenance', state: 'failed', scope: { operation: 'rotated_logs', failure: 'plan-changed' } }),
{ operation: 'rotated_logs' });
});
// #550 live test: a records operation's rows have no path, only a detail ("Prompt 12: category …").
const RECORD_JOB = Object.freeze({
id: 'a'.repeat(32), kind: 'maintenance', title: 'Pair category and tags', state: 'finished',
admission: 'records', cancellable: false, progress: 100, processed: 0, total: 0, status_line: 'Filled 2 prompts',
scope: { operation: 'pair_metadata' }, counters: { changed: 2, protected: 0, errors: 0 },
started_at: 1, finished_at: 2,
});
test('#550: path-less record rows are kept with their detail; file rows are unchanged', async () => {
const { parseDetail, filterRows } = await jobView;
const parsed = parseDetail({ ...RECORD_JOB, outcomes: { changed: [
{ path: '', root: '', root_index: null, detail: 'Prompt 2325: category “asian”; tags asian' },
{ path: '', root: '', root_index: null, detail: '' },
{ path: 'a/b.png', root: 'output', root_index: 0, detail: 'Link removed' },
] }, log: [] });
assert.deepEqual(parsed.outcomes.changed, [
{ path: '', root: '', root_index: null, detail: 'Prompt 2325: category “asian”; tags asian', record: true },
{ path: 'a/b.png', root: 'output', root_index: 0, detail: 'Link removed', record: false },
]);
// The filter reads a record's detail and a file's path.
assert.deepEqual(filterRows(parsed.outcomes.changed, 'asian').map(row => row.detail),
['Prompt 2325: category “asian”; tags asian']);
assert.deepEqual(filterRows(parsed.outcomes.changed, 'a/b').map(row => row.path), ['a/b.png']);
});
test('#550: a records operation says records, not files, for an empty group and a filter', async () => {
const { groupCopy, EMPTY_GROUP, NO_MATCH } = await jobView;
assert.deepEqual(groupCopy(RECORD_JOB), { empty: 'No records in this group.', noMatch: 'No listed record matches this filter.',
placeholder: 'Filter records…' });
for (const operation of ['duplicate_prompts', 'unused_components']) {
assert.equal(groupCopy({ ...RECORD_JOB, scope: { operation } }).empty, 'No records in this group.');
}
const files = { empty: EMPTY_GROUP, noMatch: NO_MATCH, placeholder: 'Filter by file path…' };
assert.deepEqual(groupCopy({ ...RECORD_JOB, scope: { operation: 'missing_links' } }), files);
assert.deepEqual(groupCopy({ ...RECORD_JOB, kind: 'scan', scope: {} }), files);
});
+34
View File
@@ -452,6 +452,40 @@ class TestReview(DuplicateFilesTestCase):
)
class TestReviewProtectedRows(DuplicateFilesTestCase):
async def test_cleanup_lists_each_copy_the_review_kept(self):
if not TestReview._can_symlink(self.tmpdir): # pragma: no cover
self.skipTest("symlinks are not available")
self.trio()
job, group = await self.found_group()
# b.png becomes a link after the find: the review protects it (#586).
(self.output_dir / "b.png").unlink()
os.symlink(str(self.output_dir / "a.png"), str(self.output_dir / "b.png"))
plan = await self.review(job.id, [self.keep_of(group, "a.png")])
self.assertEqual(plan["protected"]["count"], 1)
body = await self.apply(plan["token"])
cleanup = self.api._jobs.get(body["job"]["id"])
await asyncio.wait_for(asyncio.shield(cleanup.task), DISK_BOUND_SECONDS)
self.assertEqual(cleanup.state, "finished", cleanup.log)
rows = {row["path"]: row for row in cleanup.outcomes["kept"]}
self.assertEqual(
rows["b.png"],
{
"path": "b.png",
"root": rows["a.png"]["root"],
"root_index": 0,
"detail": "Kept: a link, not a regular file",
},
)
self.assertEqual(rows["a.png"]["detail"], "Kept copy")
self.assertEqual(cleanup.counters, {"removed": 1, "kept": 2, "errors": 0})
report = cleanup.report()["outcomes"]["kept"]
self.assertEqual(
sorted(row["path"] for row in report["rows"]), ["a.png", "b.png"]
)
self.assertFalse(report["truncated"])
class TestCleanupJob(DuplicateFilesTestCase):
async def test_cleanup_is_a_records_job_with_a_safety_backup(self):
self.trio()
+24
View File
@@ -266,6 +266,30 @@ class TestApply(LogCleanupTestCase):
report = json.dumps(job.report())
self.assertNotIn(self.tmpdir, report)
async def test_apply_lists_each_kept_file(self):
linked = self.seed()
plan = await self.preview()
_summary, job = await self.run_apply(plan)
expected = [
("prompt_manager.log", "Kept: the active log"),
("prompt_manager.log.9", "Kept: another file"),
("prompt_manager.log.bak", "Kept: another file"),
]
if linked:
expected.append(("prompt_manager.log.4", "Kept: a link"))
rows = job.outcomes["protected"]
self.assertEqual(
sorted((row["path"], row["detail"]) for row in rows), sorted(expected)
)
self.assertEqual(
{(row["root"], row["root_index"]) for row in rows}, {("", None)}
)
report = job.report()["outcomes"]["protected"]
self.assertEqual(
(report["count"], len(report["rows"]), report["truncated"]),
(plan["protected"]["count"], len(expected), False),
)
async def test_apply_rows_name_each_file(self):
self.seed()
plan = await self.preview()
+181
View File
@@ -0,0 +1,181 @@
"""Pair category and tags repair through the maintenance routes (#550).
``pair_metadata`` is reviewed and applied like the other maintenance plans:
``options`` must be ``{}``, the job is titled "Pair category and tags", the
safety backup follows the ``maintenance.safety_backup`` policy, and the job's
downloadable report lists every filled prompt, never cut at 200 rows.
"""
import asyncio
import os
import sys
from unittest.mock import MagicMock
_mock_server = MagicMock()
_mock_server.PromptServer.instance.routes = MagicMock()
sys.modules.setdefault("server", _mock_server)
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from py.api.jobs import MAX_OUTCOME_ROWS # noqa: E402
from py.api.maintenance import MAINTENANCE_TITLES # noqa: E402
from py.config import PromptManagerConfig # noqa: E402
try:
from tests.test_api_maintenance import WAIT, ApplyTestCase # noqa: E402
except ImportError: # discovered with tests/ as the top-level directory
from test_api_maintenance import WAIT, ApplyTestCase # noqa: E402
OPERATION = "pair_metadata"
TITLE = "Pair category and tags"
class PairMetadataAPITestCase(ApplyTestCase):
def pair(self, positive, negative="", **metadata):
pair_id = self.api.db.save_prompt_pair(positive, negative, **metadata)
with self.api.db.model.get_connection() as conn:
prompt = conn.execute(
"SELECT positive_prompt_id FROM prompt_pairs WHERE id=?", (pair_id,)
).fetchone()[0]
return pair_id, prompt
def category(self, prompt_id):
with self.api.db.model.get_connection() as conn:
return conn.execute(
"SELECT category FROM prompts WHERE id=?", (prompt_id,)
).fetchone()[0]
class TestPairMetadataRoutes(PairMetadataAPITestCase):
async def test_preview_accepts_the_operation_and_no_options(self):
self.assertEqual(MAINTENANCE_TITLES[OPERATION], TITLE)
self.pair("a cat", category="portrait", tags=["a"])
plan = await self.preview(OPERATION)
self.assertEqual(plan["operation"], OPERATION)
self.assertEqual(plan["scope"]["count"], 1)
self.assertEqual(plan["backup"]["policy"], "on")
refused = await self.preview(OPERATION, 400, options={"all": True})
self.assertEqual(refused["code"], "invalid-request")
async def test_apply_fills_with_a_backup_and_names_it(self):
_pair, prompt = self.pair("a cat", "neg", category="portrait", tags=["a", "b"])
plan, summary, job = await self.run_apply(OPERATION)
self.assertEqual(summary["title"], TITLE)
self.assertEqual(summary["admission"], "records")
self.assertEqual(job.state, "finished", job.log)
self.assertEqual(self.category(prompt), "portrait")
[backup] = self.backups()
self.assertEqual(
job.status_line,
"Filled 1 prompt from its saved pairs (1 category, 1 tag set) · "
f"Safety backup {backup}",
)
self.assertEqual(
job.outcomes["changed"][0]["detail"],
f"Prompt {prompt}: category “portrait”; tags a, b",
)
self.assertEqual(job.counters["changed"], 1)
async def test_protected_prompts_are_counted(self):
_a, prompt = self.pair("conflict", "n1", category="x")
self.pair("conflict", "n2", category="y")
self.pair("fill me", category="portrait")
plan, _summary, job = await self.run_apply(OPERATION)
self.assertEqual(plan["protected"]["reasons"], ["1 where pairs disagree"])
self.assertEqual(job.counters["protected"], 1)
self.assertIsNone(self.category(prompt))
# Live test (#550): the count comes with its rows, in the review's wording.
self.assertEqual(
job.outcomes["protected"],
[
{
"path": "",
"root": "",
"root_index": None,
"detail": (
f"Prompt {prompt}: kept, pairs disagree on category: x, y"
),
}
],
)
report = job.report()["outcomes"]["protected"]
self.assertEqual((report["count"], report["truncated"]), (1, False))
async def test_report_lists_every_kept_prompt(self):
PromptManagerConfig.SAFETY_BACKUP = False
count = MAX_OUTCOME_ROWS + 5
for n in range(count):
_pair, prompt = self.pair(f"kept {n:03d}", category="portrait")
with self.api.db.model.get_connection() as conn:
conn.execute(
"UPDATE prompts SET updated_at='2099-01-01T00:00:00+00:00' "
"WHERE id=?",
(prompt,),
)
_plan, _summary, job = await self.run_apply(OPERATION)
self.assertEqual(job.counters["protected"], count)
self.assertEqual(len(job.outcomes["protected"]), MAX_OUTCOME_ROWS)
protected = job.report()["outcomes"]["protected"]
self.assertEqual(
(len(protected["rows"]), protected["truncated"]), (count, False)
)
self.assertTrue(
protected["rows"][0]["detail"].endswith(
": kept, edited after it was saved; its empty category or tags "
"may be intentional"
)
)
async def test_policy_off_writes_without_a_backup(self):
PromptManagerConfig.SAFETY_BACKUP = False
_pair, prompt = self.pair("no backup", category="portrait")
plan, _summary, job = await self.run_apply(OPERATION)
self.assertEqual(plan["backup"]["policy"], "off")
self.assertEqual(job.state, "finished", job.log)
self.assertIn("No safety backup (the policy is off)", "\n".join(job.log))
self.assertTrue(job.status_line.endswith("· No safety backup"))
self.assertEqual(self.backups(), [])
self.assertEqual(self.category(prompt), "portrait")
async def test_reviewed_with_the_policy_on_keeps_the_backup(self):
self.pair("promise", category="portrait")
plan = await self.preview(OPERATION)
PromptManagerConfig.SAFETY_BACKUP = False
body = await self.apply(plan["token"])
job = self.api._jobs.get(body["job"]["id"])
await asyncio.wait_for(asyncio.shield(job.task), WAIT)
self.assertEqual(len(self.backups()), 1)
async def test_plan_changed_writes_nothing(self):
pair_id, prompt = self.pair("drift", category="portrait")
plan = await self.preview(OPERATION)
self.api.db.update_prompt_pair_metadata(pair_id, category="landscape")
body = await self.apply(plan["token"])
job = self.api._jobs.get(body["job"]["id"])
await asyncio.wait_for(asyncio.shield(job.task), WAIT)
self.assertEqual(job.state, "failed")
self.assertEqual(job.last_event["code"], "plan-changed")
self.assertIsNone(self.category(prompt))
async def test_report_lists_every_filled_prompt(self):
PromptManagerConfig.SAFETY_BACKUP = False
count = MAX_OUTCOME_ROWS + 5
for n in range(count):
self.pair(f"bulk {n:03d}", category="portrait")
_plan, _summary, job = await self.run_apply(OPERATION)
self.assertEqual(job.state, "finished", job.log)
self.assertEqual(len(job.outcomes["changed"]), MAX_OUTCOME_ROWS)
response = await self.client.request(
"GET", f"/prompt_manager/jobs/{job.id}/report"
)
self.assertEqual(response.status, 200)
report = await response.json()
changed = report["outcomes"]["changed"]
self.assertEqual(changed["count"], count)
self.assertEqual(len(changed["rows"]), count)
self.assertFalse(changed["truncated"])
self.assertTrue(
all(
row["detail"].endswith("category “portrait”") for row in changed["rows"]
)
)
@@ -0,0 +1,140 @@
"""Protected maintenance items are listed in Activity, not only counted (#586).
Each record a maintenance job keeps gets one ``protected`` row with its
reason, in the wording of that operation's review, and the job's
downloadable report lists every one of them (never cut at 200 rows). A kept
image link names its file, so Activity can still open and locate it.
"""
import os
import shutil
import sys
from pathlib import Path
from unittest.mock import MagicMock
_mock_server = MagicMock()
_mock_server.PromptServer.instance.routes = MagicMock()
sys.modules.setdefault("server", _mock_server)
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from py.api.jobs import MAX_OUTCOME_ROWS # noqa: E402
from py.config import GalleryConfig, PromptManagerConfig # noqa: E402
try:
from tests.test_api_maintenance import ApplyTestCase # noqa: E402
except ImportError: # discovered with tests/ as the top-level directory
from test_api_maintenance import ApplyTestCase # noqa: E402
def _record(detail):
return {"path": "", "root": "", "root_index": None, "detail": detail}
class TestProtectedRows(ApplyTestCase):
def duplicate(self, text, created_at="2999-01-01"):
with self.api.db.model.get_connection() as conn:
return conn.execute(
"INSERT INTO prompts(text, created_at) VALUES(?, ?)",
(text.upper(), created_at),
).lastrowid
def assert_complete(self, job, count):
self.assertEqual(job.counters["protected"], count)
report = job.report()["outcomes"]["protected"]
self.assertEqual(
(report["count"], len(report["rows"]), report["truncated"]),
(count, count, False),
)
async def test_duplicate_prompts_lists_each_kept_copy(self):
paired = self.save("A dog")
self.api.db.save_prompt_pair("A dog", "")
paired_copy = self.duplicate("A dog")
titled = self.save("A bird")
titled_copy = self.duplicate("A bird")
with self.api.db.model.get_connection() as conn:
conn.execute(
"INSERT INTO prompt_library_state(prompt_id,title) VALUES(?,?)",
(titled_copy, "Bird recipe"),
)
plan, _summary, job = await self.run_apply("duplicate_prompts")
self.assertEqual(job.state, "finished", job.log)
self.assertEqual(
sorted(row["detail"] for row in job.outcomes["protected"]),
sorted(
[
f"Prompt {paired}: kept, belongs to a saved pair",
f"Prompt {paired_copy}: kept, belongs to a saved pair",
f"Prompt {titled}: kept, a copy has its own Library title",
f"Prompt {titled_copy}: kept, a copy has its own Library title",
]
),
)
self.assertEqual(plan["protected"]["count"], 4)
self.assert_complete(job, 4)
async def test_unused_components_lists_each_kept_prompt_with_its_reasons(self):
self.save("Orphan")
tagged = self.save("Kept", tags=["__protected__"])
self.api.db.save_prompt_pair("Paired", "")
paired = self.api.db.search_prompts(text="Paired")[0]["id"]
with self.api.db.model.get_connection() as conn:
conn.execute(
"INSERT INTO prompt_tags(prompt_id, tag_id) "
"SELECT ?, id FROM tags WHERE name='__protected__'",
(paired,),
)
plan, _summary, job = await self.run_apply("unused_components")
self.assertEqual(job.state, "finished", job.log)
self.assertEqual(
sorted(job.outcomes["protected"], key=lambda row: row["detail"]),
sorted(
[
_record(f"Prompt {tagged}: kept, tagged __protected__"),
_record(
f"Prompt {paired}: kept, in a saved pair; "
"tagged __protected__"
),
],
key=lambda row: row["detail"],
),
)
self.assertEqual(plan["protected"]["count"], 2)
self.assert_complete(job, 2)
async def test_report_lists_every_kept_prompt(self):
PromptManagerConfig.SAFETY_BACKUP = False
count = MAX_OUTCOME_ROWS + 5
for n in range(count):
self.save(f"Kept {n:03d}", tags=["__protected__"])
_plan, _summary, job = await self.run_apply("unused_components")
self.assertEqual(len(job.outcomes["protected"]), MAX_OUTCOME_ROWS)
self.assert_complete(job, count)
async def test_missing_links_lists_each_kept_link_with_its_file(self):
unmounted = Path(self.tmpdir) / "unmounted"
unmounted.mkdir()
GalleryConfig.MONITORING_DIRECTORIES = [str(self.output_dir), str(unmounted)]
prompt = self.save("Links")
self.image(prompt, "gone.png", exists=False)
self.image(prompt, "sub/away.png", folder=unmounted, exists=False)
self.image(prompt, "x.png", folder=Path(self.tmpdir) / "elsewhere")
shutil.rmtree(unmounted)
plan, _summary, job = await self.run_apply("missing_links")
self.assertEqual(job.state, "finished", job.log)
self.assertEqual(plan["protected"]["count"], 2)
rows = sorted(job.outcomes["protected"], key=lambda row: row["path"])
self.assertEqual(
[(row["path"], row["detail"]) for row in rows],
[
("sub/away.png", "Kept: link under an unavailable source"),
("x.png", "Kept: link outside the image sources"),
],
)
# The gone source keeps the path under its label; outside every
# source the row names the file only.
self.assertEqual((rows[0]["root"], rows[0]["root_index"]), ("unmounted", None))
self.assertEqual((rows[1]["root"], rows[1]["root_index"]), ("", None))
self.assertNotIn(self.tmpdir, str(job.report()))
self.assert_complete(job, 2)
+9 -2
View File
@@ -363,11 +363,11 @@ class TestWorkflowStorageOutcomes(WorkflowStorageTestCase):
prompt_id = self.old_library(count=4)
conn = self.api.db.model.get_connection()
with conn:
conn.execute(
image_id = conn.execute(
"INSERT INTO generated_images (prompt_id, image_path, filename, "
"file_path, workflow_data) VALUES (?, '/x/b', 'b', '/x/b', x'00')",
(prompt_id,),
)
).lastrowid
store_text(conn, "a document no record references")
_plan, _summary, job = await self.start()
await self.finish(job)
@@ -377,6 +377,13 @@ class TestWorkflowStorageOutcomes(WorkflowStorageTestCase):
"Rewrote 4 image records; 1 stored value not text, kept as is", log
)
self.assertIn("Removed 1 unused workflow document", log)
# #586: the kept value has its row, in the review's wording.
self.assertEqual(
[row["detail"] for row in job.outcomes["protected"]],
[f"Image record {image_id}: workflow, stored value not text; kept as is"],
)
report = job.report()["outcomes"]["protected"]
self.assertEqual((report["count"], report["truncated"]), (1, False))
async def reclaim_with(self, **patches):
self.old_library(count=4)
+5 -5
View File
@@ -908,20 +908,20 @@ class TestWorkspaceImages(ImageAPITestCase):
async def test_one_cards_parse_error_spares_the_page(self):
prompt = self._save_prompt("a fox")
for name, mtime in (("bad.png", 200), ("ok.png", 100)):
# Cards are read on the image pool (#523), in no fixed order: the
# parser fails on bad.png's own record (seed 6), not on a call count.
for name, mtime, seed in (("bad.png", 200, 6), ("ok.png", 100, 5)):
path = records_png(
self.output_dir / name, seeded_records("a fox", 5), mtime
self.output_dir / name, seeded_records("a fox", seed), mtime
)
self.api.db.link_image_to_prompt(prompt, str(path))
real_parse = header_params_of_images()
def parsing(info):
if parsing.calls == 0:
parsing.calls += 1
if '"seed": 6' in info["prompt"]:
raise RuntimeError("unexpected parser bug")
return real_parse(info)
parsing.calls = 0
with (
patch("py.api.images.header_params", side_effect=parsing),
self.assertLogs(self.api.logger, "WARNING") as logged,
+12
View File
@@ -146,6 +146,18 @@ class TestPreviewQueueRefusedAtShutdown(unittest.TestCase):
class TestMapParallelRefusedAtShutdown(unittest.TestCase):
def setUp(self):
# The job pool is long-lived (#523): start each test without one, so
# the patched pool class is the one created.
self.assertTrue(parallel.shutdown_job_pool(30))
self.addCleanup(parallel.shutdown_job_pool, 30)
def test_the_refused_pool_is_not_reused(self):
with mock.patch.object(parallel, "ThreadPoolExecutor", _ShutPool):
parallel.map_parallel(lambda n: n, [1, 2], 2)
self.assertFalse(parallel.job_pool_started())
self.assertEqual(parallel.map_parallel(lambda n: n + 1, [1, 2], 2), [2, 3])
def test_a_shut_pool_runs_the_items_on_the_calling_thread(self):
records = _capture(self, "prompt_manager.parallel")
with mock.patch.object(parallel, "ThreadPoolExecutor", _ShutPool):
+241
View File
@@ -0,0 +1,241 @@
"""Image work runs on PromptManager's own pools, never the default executor (#523).
glibc keeps each thread's malloc arena at its high-water mark, so image work
spread over the default executor's threads (min(32, cpus + 4)) kept an
image-sized buffer per thread. Interactive image work (card facts, inspector
reads, view previews) runs on the fixed interactive image pool
(``pm-image``); jobs (scan metadata, thumbnail jobs, capture previews) on the
long-lived Worker threads pool (``pm-worker``). After a scan or thumbnail
job, glibc's free pages are released once (``malloc_trim``).
"""
import asyncio
import threading
from unittest.mock import patch
import py.api as api_package
import py.api.images as images_module
import py.api.inspection_service as inspection_service
import py.api.thumbnail_auto as auto
from utils import parallel
try:
from tests.test_api_images import make_png
from tests.test_thumbnail_auto import WAIT, AutoThumbnailCase, _entry
except ImportError: # discovered with tests/ as the top-level directory
from test_api_images import make_png
from test_thumbnail_auto import WAIT, AutoThumbnailCase, _entry
IMAGE_THREAD = auto.POOL_PREFIX + "_"
JOB_THREAD = parallel.JOB_THREAD_PREFIX + "_"
class ImagePoolCase(AutoThumbnailCase):
async def asyncSetUp(self):
await super().asyncSetUp()
self.assertTrue(parallel.shutdown_job_pool(WAIT))
self.addCleanup(parallel.shutdown_job_pool, WAIT)
self.threads = []
self.lock = threading.Lock()
def spy(self, owner, name):
"""Record the thread every call of ``owner.name`` runs on."""
original = getattr(owner, name)
def recording(*args, **kwargs):
with self.lock:
self.threads.append(threading.current_thread().name)
return original(*args, **kwargs)
patcher = patch.object(owner, name, recording)
patcher.start()
self.addCleanup(patcher.stop)
def assert_threads(self, prefix):
self.assertTrue(self.threads, "the spied call never ran")
self.assertTrue(
all(name.startswith(prefix) for name in self.threads), self.threads
)
async def run_job(self, path, body=None):
response = await self.client.post(path, json=body or {})
self.assertEqual(response.status, 202, await response.text())
job = self.api._jobs.get((await response.json())["job"]["id"])
await asyncio.wait_for(asyncio.shield(job.task), WAIT)
return job
class TestInteractiveImagePool(ImagePoolCase):
async def test_workspace_card_facts_are_read_on_the_image_pool(self):
for index in range(5):
make_png(self.output_dir / f"c{index}.png", size=(32, 32))
self.spy(images_module, "_card_header")
response = await self.client.get(
"/prompt_manager/images/output?mode=workspace&limit=100"
)
self.assertEqual(response.status, 200, await response.text())
images = (await response.json())["images"]
self.assertEqual([image["width"] for image in images], [32] * 5)
self.assert_threads(IMAGE_THREAD)
self.assertEqual(len(self.threads), 5)
async def test_cached_card_facts_need_no_image_pool_trip(self):
make_png(self.output_dir / "c.png", size=(32, 32))
url = "/prompt_manager/images/output?mode=workspace&limit=100"
await (await self.client.get(url)).read()
self.spy(images_module, "_card_header")
response = await self.client.get(url)
self.assertEqual((await response.json())["images"][0]["width"], 32)
self.assertEqual(self.threads, [])
async def test_the_inspector_reads_on_the_image_pool(self):
make_png(self.output_dir / "i.png", size=(32, 32))
self.spy(inspection_service, "read_inspection_file")
response = await self.client.get("/prompt_manager/images/inspect/i.png?root=0")
self.assertEqual(response.status, 200, await response.text())
self.assert_threads(IMAGE_THREAD)
async def test_view_previews_and_card_reads_share_three_threads(self):
names = [f"v{i}.png" for i in range(auto.ON_DEMAND_LIMIT + 3)]
for name in names:
make_png(self.output_dir / name, size=(64, 64))
calls = []
original = self.api._generate_one
def recording(*args, **kwargs):
calls.append(threading.current_thread().name)
return original(*args, **kwargs)
self.api._generate_one = recording
for name in names:
url = _entry(self.api, self.output_dir, name)["thumbnail_url"]
await (await self.client.get(url)).read()
self.assertEqual(len(calls), len(names))
self.assertTrue(all(name.startswith(IMAGE_THREAD) for name in calls), calls)
self.assertLessEqual(len(self.api._image_pool_threads()), auto.IMAGE_POOL_SIZE)
def test_map_image_keeps_order_and_runs_inline_on_a_pool_thread(self):
self.assertEqual(self.api._map_image(lambda n: n * 2, [3, 1, 2]), [6, 2, 4])
nested = self.api._image_pool().submit(
self.api._map_image, lambda _: threading.current_thread().name, [1]
)
(name,) = nested.result(WAIT)
self.assertTrue(name.startswith(IMAGE_THREAD), name)
def test_map_image_after_a_close_runs_on_the_calling_thread(self):
pool = self.api._image_pool()
pool.shutdown(wait=True)
here = threading.current_thread().name
with patch.object(self.api, "_image_pool", return_value=pool):
names = self.api._map_image(
lambda _: threading.current_thread().name, [1, 2]
)
self.assertEqual(names, [here, here])
async def test_after_the_final_close_no_image_thread_starts(self):
await asyncio.get_running_loop().run_in_executor(
None, lambda: self.api.close_auto_thumbnails(final=True)
)
self.assertIsNone(self.api._image_pool())
here = threading.current_thread().name
names = self.api._map_image(lambda _: threading.current_thread().name, [1])
self.assertEqual(names, [here])
name = await self.api._run_image(lambda: threading.current_thread().name)
self.assertFalse(name.startswith(IMAGE_THREAD), name)
self.assertEqual(self.api._image_pool_threads(), [])
def test_map_image_runs_items_a_shutdown_cancelled_here(self):
entered = threading.Barrier(auto.IMAGE_POOL_SIZE + 1)
release = threading.Event()
self.addCleanup(release.set)
pool = self.api._image_pool()
for _ in range(auto.IMAGE_POOL_SIZE): # occupy every pool thread
pool.submit(lambda: (entered.wait(WAIT), release.wait(WAIT)))
entered.wait(WAIT)
here = threading.current_thread().name
results, submitted = {}, threading.Event()
class Counting:
calls = 0
def submit(self, *args):
future = pool.submit(*args)
Counting.calls += 1
if Counting.calls == 2:
submitted.set()
return future
def call():
with patch.object(self.api, "_image_pool", return_value=Counting()):
results["names"] = self.api._map_image(
lambda _: threading.current_thread().name, [1, 2]
)
caller = threading.Thread(target=call, name="map-image-caller")
caller.start()
# The caller's two items are queued behind the busy threads; the
# shutdown cancels them and the caller runs them itself.
self.assertTrue(submitted.wait(WAIT))
pool.shutdown(wait=False, cancel_futures=True)
caller.join(WAIT)
release.set()
self.assertFalse(caller.is_alive())
self.assertEqual(results["names"], ["map-image-caller"] * 2)
self.assertNotEqual(here, "map-image-caller")
async def test_run_image_after_a_close_uses_the_default_executor(self):
pool = self.api._image_pool()
pool.shutdown(wait=True)
with patch.object(self.api, "_image_pool", return_value=pool):
name = await self.api._run_image(lambda: threading.current_thread().name)
self.assertFalse(name.startswith(IMAGE_THREAD), name)
class TestJobPool(ImagePoolCase):
def test_scan_metadata_is_read_on_the_job_pool(self):
paths = [
make_png(self.output_dir / f"s{i}.png", size=(16, 16)) for i in range(4)
]
self.spy(self.api, "_extract_comfyui_metadata")
results = self.api._extract_batch_metadata_sync(paths)
self.assertEqual([path for path, _ in results], paths)
self.assert_threads(JOB_THREAD)
async def test_thumbnail_job_writes_on_the_job_pool_and_trims_after(self):
for index in range(3):
make_png(self.output_dir / f"t{index}.png", size=(64, 64))
self.spy(self.api, "_generate_one")
with patch.object(api_package, "release_free_memory") as trim:
job = await self.run_job("/prompt_manager/jobs/thumbnails")
self.assertEqual(job.state, "finished")
self.assert_threads(JOB_THREAD)
trim.assert_called_once_with()
async def test_scan_job_trims_after_it_finishes(self):
make_png(self.output_dir / "s.png", size=(16, 16))
states = []
def trim():
states.append(self.api._jobs.active("scan"))
return True
with patch.object(api_package, "release_free_memory", side_effect=trim):
job = await self.run_job("/prompt_manager/jobs/scan")
self.assertEqual(job.state, "finished")
self.assertGreaterEqual(len(states), 1)
self.assertIsNone(states[0], "trimmed after the scan finished")
async def test_a_failing_trim_never_fails_the_job(self):
make_png(self.output_dir / "t.png", size=(64, 64))
with patch.object(
api_package, "release_free_memory", side_effect=RuntimeError("no")
):
job = await self.run_job("/prompt_manager/jobs/thumbnails")
self.assertEqual(job.state, "finished")
async def test_capture_previews_write_on_the_job_pool(self):
path = make_png(self.output_dir / "a.png", size=(64, 64))
self.spy(self.api, "_generate_one")
self.assertEqual(self.api.queue_capture_thumbnail(str(path)), "queued")
await self._idle()
self.assert_threads(JOB_THREAD)
+67
View File
@@ -136,6 +136,11 @@ def cleanup_package(package, require_stopped=True):
if registration is not None:
assert registration.dispose(3), "package history worker leaked"
close_every_database()
# The package's exit actions outlive it otherwise: at interpreter exit
# they would import from a package no longer in sys.modules (#580).
exit_signal = getattr(package, "_exit_signal", None)
if exit_signal is not None:
exit_signal.reset()
sys.modules.pop(package.__name__, None)
@@ -402,6 +407,29 @@ class TestConsoleBanner(unittest.TestCase):
self.assertNotIn("Loaded:", banner)
class TestFixtureExitActions(unittest.TestCase):
"""A disposed fixture package runs nothing at interpreter exit (#580)."""
def test_a_disposed_package_leaves_no_exit_action(self):
tmp = tempfile.TemporaryDirectory()
self.addCleanup(tmp.cleanup)
self.addCleanup(close_every_database)
output_dir = os.path.join(tmp.name, "output")
os.makedirs(output_dir)
name = "prompt_manager_pkg_exit_actions"
self.addCleanup(sys.modules.pop, name, None)
with contextlib.redirect_stdout(io.StringIO()):
with load_package(name, os.path.join(tmp.name, "p.db"), output_dir) as (
package,
_,
):
signal = package._exit_signal
self.assertTrue(signal._on_exit, "the package installs exit actions")
self.assertEqual(signal._on_exit, [])
class PriorFileHandler(logging.FileHandler):
"""A caller-owned sink that stays closed once its owner closes it.
@@ -771,6 +799,45 @@ class TestPackageTeardown(PackageTeardownCase):
self.assertIsNone(seen["worker"])
self.assertTrue(all(result.values()), result)
def test_teardown_stops_the_job_pool_once_a_job_used_it(self):
"""#523: the long-lived job pool is a teardown part of its own."""
package = self.load()
parallel = sys.modules[package.__name__ + ".utils.parallel"]
self.assertEqual(parallel.map_parallel(lambda n: n * 2, [1, 2], 2), [2, 4])
# This package's own pool (the suite's top-level one may also run).
threads = list(parallel._pool[1]._threads)
self.assertTrue(threads)
result = package._stop_background_work()
self.assertEqual(list(result), ["monitor", "thumbnails", "workers", "tracker"])
self.assertTrue(all(result.values()), result)
self.assertFalse(parallel.job_pool_started())
self.assertFalse(any(t.is_alive() for t in threads))
def test_teardown_reports_a_job_thread_past_its_bound(self):
package = self.load(require_stopped=False)
parallel = sys.modules[package.__name__ + ".utils.parallel"]
entered, release = threading.Event(), threading.Event()
def hold(_):
entered.set()
release.wait(30)
caller = threading.Thread(target=parallel.map_parallel, args=(hold, [1], 1))
caller.start()
self.addCleanup(caller.join, 30)
self.addCleanup(release.set)
self.assertTrue(entered.wait(30))
with (
mock.patch.object(parallel, "JOB_POOL_JOIN_SECONDS", 0.05),
self.assertLogs("prompt_manager.init", "WARNING") as logs,
):
result = package._stop_background_work()
self.assertFalse(result["workers"])
self.assertTrue(result["tracker"])
self.assertRegex("\n".join(logs.output), r"not stopped in time .*workers")
def test_teardown_kills_ffmpeg_and_stops_the_preview_worker(self):
package = self.load()
held = self.hold_a_preview(package, advance=4.0, kill_releases=True)
+30
View File
@@ -490,6 +490,36 @@ class TestJobOutcomesAndLog(unittest.TestCase):
with self.assertRaises(ValueError):
job.record_counted("bogus", 1)
def test_record_counted_complete_keeps_every_row_for_the_report(self):
# #550: a repair's record of each change is never cut at the cap.
job = Job("maintenance", "Pair category and tags", groups=("changed", "errors"))
rows = [
{"path": "", "detail": f"Prompt {i}"} for i in range(MAX_OUTCOME_ROWS + 5)
]
job.record_counted("changed", len(rows), rows, complete=True)
self.assertEqual(len(job.outcomes["changed"]), MAX_OUTCOME_ROWS)
report = job.report()["outcomes"]
self.assertEqual(len(report["changed"]["rows"]), MAX_OUTCOME_ROWS + 5)
self.assertEqual(
report["changed"]["rows"][-1]["detail"], f"Prompt {MAX_OUTCOME_ROWS + 4}"
)
self.assertFalse(report["changed"]["truncated"])
self.assertEqual(report["errors"], {"count": 0, "rows": [], "truncated": False})
def test_record_into_a_complete_group_keeps_the_row_for_the_report(self):
# #586: a group kept whole stays whole when later rows are recorded.
job = Job("duplicate_cleanup", "Remove", groups=("kept",))
rows = [
{"path": f"f{i}.png", "detail": "Kept"} for i in range(MAX_OUTCOME_ROWS)
]
job.record_counted("kept", len(rows), rows, complete=True)
job.record("kept", {"path": "late.png", "detail": "Kept copy"}, "= late.png")
self.assertEqual(len(job.outcomes["kept"]), MAX_OUTCOME_ROWS)
report = job.report()["outcomes"]["kept"]
self.assertEqual(report["count"], MAX_OUTCOME_ROWS + 1)
self.assertEqual(report["rows"][-1]["path"], "late.png")
self.assertFalse(report["truncated"])
def test_thumbnail_job_groups(self):
self.assertEqual(
jobs_module.THUMBNAIL_GROUP_NAMES, ("generated", "skipped", "errors")
+539
View File
@@ -0,0 +1,539 @@
"""Pair category and tags repair (#550): the ``pair_metadata`` maintenance plan.
Before #546 a Pair node run saved its category and tags on the saved pair
only, so the positive prompt it created stayed Uncategorized and untagged.
``pair_metadata`` reviews those prompts and fills only their empty fields
from the prompt's non-archived pairs when every pair with a value agrees.
It never touches a negative, a saved pair or ``updated_at``, and a prompt
the Pair node did not create, that was edited, archived, or whose pairs
disagree, is kept (``550-design.md``).
"""
import os
import sys
import unittest
from unittest.mock import patch
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from database import maintenance # noqa: E402
from database import pair_metadata_repair as repair # noqa: E402
try:
from tests.test_maintenance_plans import MaintenanceDatabaseCase # noqa: E402
except ImportError: # discovered with tests/ as the top-level directory
from test_maintenance_plans import MaintenanceDatabaseCase # noqa: E402
OPERATION = "pair_metadata"
class PairMetadataCase(MaintenanceDatabaseCase):
def pair(self, positive, negative="", **metadata):
"""A pre-#546 Pair save: the pair gets the metadata, the prompt none."""
pair_id = self.db.save_prompt_pair(positive, negative, **metadata)
return pair_id, self.positive_of(pair_id)
def positive_of(self, pair_id):
return self.sql(
"SELECT positive_prompt_id FROM prompt_pairs WHERE id=?", pair_id
)[0][0]
def sql(self, statement, *args):
with self.db.model.get_connection() as conn:
return [tuple(row) for row in conn.execute(statement, args)]
def run_sql(self, statement, *args):
with self.db.model.get_connection() as conn:
conn.execute(statement, args)
def category(self, prompt_id):
return self.sql("SELECT category FROM prompts WHERE id=?", prompt_id)[0][0]
def tags(self, prompt_id):
return sorted(
row[0]
for row in self.sql(
"SELECT t.name FROM prompt_tags pt JOIN tags t ON t.id=pt.tag_id "
"WHERE pt.prompt_id=?",
prompt_id,
)
)
def preview(self, operation=OPERATION, roots=None):
return self.db.preview_maintenance(operation, [])
def apply(self, draft):
return self.db.apply_maintenance(
draft.operation, draft.fingerprint, draft.targets, []
)
def edit_stamp(self, prompt_id, updated_at="2099-01-01T00:00:00+00:00"):
self.run_sql(
"UPDATE prompts SET updated_at=? WHERE id=?", updated_at, prompt_id
)
def details(self, draft):
return [(row["label"], row["detail"]) for row in draft.details]
class TestClassification(PairMetadataCase):
def test_fills_empty_category_and_tags_from_the_pair(self):
pair_id, prompt = self.pair(
"a cat on a sofa", "blurry", category="portrait", tags=["b", "a"]
)
draft = self.preview()
self.assertEqual(draft.operation, OPERATION)
self.assertEqual(draft.targets, [[prompt, True, True]])
self.assertEqual(draft.scope["count"], 1)
self.assertEqual(
draft.scope["label"], "1 prompt to fill: 1 category, 1 tag set"
)
self.assertEqual(draft.protected, {"count": 0, "reasons": []})
self.assertEqual(
draft.media,
{
"deletes_files": False,
"label": "No image file or saved pair is changed.",
},
)
self.assertIn(
(
"a cat on a sofa",
f"Fills category “portrait” and 2 tags: a, b (pair {pair_id})",
),
self.details(draft),
)
self.assertEqual(draft.details[-1]["label"], "Record")
self.assertEqual(draft.details[-1]["detail"], repair.RECORD_NOTE)
def test_empty_plan(self):
self.save("plain prompt")
draft = self.preview()
self.assertEqual(draft.targets, [])
self.assertEqual(draft.scope["count"], 0)
self.assertEqual(
draft.scope["label"], "No Pair prompts are missing a category or tags"
)
self.assertEqual(draft.details, [])
def test_prompt_already_complete_is_not_counted(self):
_pair, prompt = self.pair("done", category="portrait", tags=["a"])
self.run_sql("UPDATE prompts SET category='portrait' WHERE id=?", prompt)
self.db.set_prompt_tags(prompt, ["a"])
self.edit_stamp(
prompt, self.sql("SELECT created_at FROM prompts WHERE id=?", prompt)[0][0]
)
self.assertEqual(self.preview().scope["count"], 0)
def test_whitespace_category_is_blank_on_prompt_and_pair(self):
_p1, blank_prompt = self.pair("ws prompt", category=" portrait ")
self.run_sql("UPDATE prompts SET category=' ' WHERE id=?", blank_prompt)
_p2, blank_pair = self.pair("ws pair", category=" ")
draft = self.preview()
self.assertEqual(draft.targets, [[blank_prompt, True, False]])
self.assertIn("“portrait”", self.details(draft)[0][1])
self.assertNotIn(blank_pair, [t[0] for t in draft.targets])
def test_mixed_fields_category_fills_while_tags_conflict(self):
_a, prompt = self.pair("mixed", "n1", category="portrait", tags=["a"])
self.pair("mixed", "n2", category="portrait", tags=["b"])
draft = self.preview()
self.assertEqual(draft.targets, [[prompt, True, False]])
self.assertEqual(draft.scope["count"], 1)
self.assertEqual(draft.protected["count"], 1)
self.assertEqual(draft.protected["reasons"], ["1 where pairs disagree"])
self.assertIn(("mixed", "Kept: pairs disagree on tags"), self.details(draft))
def test_mixed_fields_tags_fill_while_category_conflicts(self):
_a, prompt = self.pair("mixed2", "n1", category="portrait", tags=["a"])
self.pair("mixed2", "n2", category="landscape", tags=["a"])
draft = self.preview()
self.assertEqual(draft.targets, [[prompt, False, True]])
self.assertIn(
("mixed2", "Kept: pairs disagree on category: landscape, portrait"),
self.details(draft),
)
def test_category_set_and_tags_empty_fills_tags_only(self):
_a, prompt = self.pair("has cat", category="portrait", tags=["a"])
self.run_sql("UPDATE prompts SET category='mine' WHERE id=?", prompt)
draft = self.preview()
self.assertEqual(draft.targets, [[prompt, False, True]])
def test_pairs_without_a_value_are_ignored(self):
_a, prompt = self.pair("partial", "n1", category="portrait", tags=["a"])
self.pair("partial", "n2") # no category, no tags
self.assertEqual(self.preview().targets, [[prompt, True, True]])
def test_equal_values_across_pairs_fill(self):
_a, prompt = self.pair("same", "n1", category="portrait", tags=["a", "b"])
self.pair("same", "n2", category="portrait", tags=["b", "a"])
self.assertEqual(self.preview().targets, [[prompt, True, True]])
def test_category_is_case_sensitive(self):
_a, prompt = self.pair("case", "n1", category="Portrait")
self.pair("case", "n2", category="portrait")
draft = self.preview()
self.assertEqual(draft.targets, [])
self.assertEqual(draft.protected["count"], 1)
def test_edited_prompt_is_protected(self):
cases = {
"later": "2099-01-01T00:00:00+00:00",
"null": None,
"legacy": "2020-01-01 10:00:00",
"garbage": "yesterday",
}
for name, stamp in cases.items():
with self.subTest(name=name):
_a, prompt = self.pair(f"edited {name}", category="portrait")
self.edit_stamp(prompt, stamp)
draft = self.preview()
self.assertNotIn(prompt, [t[0] for t in draft.targets])
self.assertIn(
(
f"edited {name}",
"Kept: edited after it was saved; its empty category or "
"tags may be intentional",
),
self.details(draft),
)
self.run_sql(
"DELETE FROM prompt_pairs WHERE positive_prompt_id=?", prompt
)
def test_first_release_stamps_microseconds_apart_are_never_edited(self):
_a, prompt = self.pair("caa9649f", category="portrait")
self.run_sql(
"UPDATE prompts SET created_at=?, updated_at=? WHERE id=?",
"2026-10-18T10:00:00.000100+00:00",
"2026-10-18T10:00:00.000180+00:00",
prompt,
)
self.run_sql(
"UPDATE prompt_pairs SET created_at=? WHERE positive_prompt_id=?",
"2026-10-18T10:00:00.000400+00:00",
prompt,
)
self.assertEqual(self.preview().targets, [[prompt, True, False]])
def test_edited_beats_conflict(self):
_a, prompt = self.pair("both", "n1", category="portrait")
self.pair("both", "n2", category="landscape")
self.edit_stamp(prompt)
draft = self.preview()
self.assertEqual(draft.protected["reasons"], ["1 edited after it was saved"])
def test_archived_prompt_is_protected(self):
_a, prompt = self.pair("archived", category="portrait")
self.run_sql(
"INSERT INTO prompt_archive(prompt_id, archived_at) VALUES(?, ?)",
prompt,
"2026-10-31T00:00:00+00:00",
)
draft = self.preview()
self.assertEqual(draft.targets, [])
self.assertEqual(draft.protected["reasons"], ["1 archived in the Library"])
self.assertIn(
("archived", "Kept: archived in the Library"), self.details(draft)
)
def test_archived_pairs_are_no_source(self):
pair_id, prompt = self.pair("old pair", "n1", category="portrait")
self.db.archive_prompt_pair(pair_id)
draft = self.preview()
self.assertEqual(draft.targets, [])
self.assertEqual(draft.protected["count"], 0)
# An archived pair also never disagrees with a live one.
self.pair("old pair", "n2", category="landscape")
self.assertEqual(self.preview().targets, [[prompt, True, False]])
self.assertIn("landscape", self.details(self.preview())[0][1])
def test_prompt_not_created_by_its_pair_is_out_of_scope(self):
prompt = self.save("saved by the node first")
self.run_sql(
"UPDATE prompts SET created_at=?, updated_at=? WHERE id=?",
"2026-01-01T00:00:00+00:00",
"2026-01-01T00:00:00+00:00",
prompt,
)
self.db.save_prompt_pair("saved by the node first", category="portrait")
draft = self.preview()
self.assertEqual(draft.targets, [])
self.assertEqual(draft.protected["count"], 0)
def test_negative_is_never_a_target_and_positive_role_sources_only(self):
_a, both = self.pair("dual", "shared negative", category="portrait")
neg = self.sql("SELECT id FROM prompts WHERE text='shared negative'")[0][0]
# The negative of pair A is the positive of pair B, made later: not
# Pair-created as a positive, so out of scope.
self.db.save_prompt_pair("shared negative", category="landscape")
draft = self.preview()
self.assertEqual(draft.targets, [[both, True, False]])
self.assertNotIn(neg, [t[0] for t in draft.targets])
def test_positive_first_then_negative_is_filled_from_positive_pairs_only(self):
_a, prompt = self.pair("pos first", category="portrait")
self.db.save_prompt_pair("another", "pos first", category="landscape")
draft = self.preview()
self.assertIn([prompt, True, False], draft.targets)
self.assertIn("“portrait”", dict(self.details(draft))["pos first"])
def test_more_than_twenty_tags_is_protected(self):
tags = [f"t{n:02d}" for n in range(21)]
_a, prompt = self.pair("many tags", category="portrait", tags=tags)
draft = self.preview()
self.assertEqual(draft.targets, [[prompt, True, False]])
self.assertEqual(
draft.protected["reasons"], ["1 whose pair has more than 20 tags"]
)
self.assertIn(
("many tags", "Kept: its pair has more than 20 tags"), self.details(draft)
)
def test_protected_rows_come_first_within_fifty(self):
_c, conflict = self.pair("zz conflict", "n1", category="x")
self.pair("zz conflict", "n2", category="y")
for n in range(60):
self.pair(f"fill {n:02d}", category="portrait")
draft = self.preview()
self.assertEqual(len(draft.details), maintenance.MAX_DETAILS)
self.assertEqual(draft.details[0]["label"], "zz conflict")
self.assertEqual(draft.details[-1]["label"], "Record")
self.assertEqual(draft.scope["count"], 60)
def test_detail_rows_list_conflicts_before_other_kept_prompts(self):
_a, archived = self.pair("aa archived", category="portrait")
self.run_sql(
"INSERT INTO prompt_archive(prompt_id, archived_at) VALUES(?, ?)",
archived,
"2026-10-31T00:00:00+00:00",
)
_b, edited = self.pair("bb edited", category="portrait")
self.edit_stamp(edited)
self.pair("zz conflict", "n1", category="x")
self.pair("zz conflict", "n2", category="y")
draft = self.preview()
self.assertEqual(
[label for label, _detail in self.details(draft)],
["zz conflict", "bb edited", "aa archived"],
)
# The reasons keep the precedence order.
self.assertEqual(
draft.protected["reasons"],
[
"1 archived in the Library",
"1 edited after it was saved",
"1 where pairs disagree",
],
)
def test_same_library_same_fingerprint_and_pair_use_does_not_change_it(self):
pair_id, _prompt = self.pair("fp", category="portrait", tags=["a"])
first = self.preview().fingerprint
self.db.record_prompt_pair_use(pair_id)
self.db.update_prompt_pair_metadata(pair_id, notes="a note", title="t")
self.assertEqual(self.preview().fingerprint, first)
class TestPreviewWritesNothing(PairMetadataCase):
def test_preview_reads_one_read_only_snapshot(self):
self.pair("cat", category="portrait", tags=["a"])
before = self.sql("SELECT * FROM prompts ORDER BY id")
tags = self.sql("SELECT * FROM prompt_tags ORDER BY prompt_id, tag_id")
connect = maintenance.sqlite3.connect
opened = []
def spy(*args, **kwargs):
opened.append(args[0])
return connect(*args, **kwargs)
with patch.object(maintenance.sqlite3, "connect", side_effect=spy):
self.preview()
self.assertEqual(len(opened), 1)
self.assertTrue(str(opened[0]).endswith("?mode=ro"))
self.assertEqual(self.sql("SELECT * FROM prompts ORDER BY id"), before)
self.assertEqual(
self.sql("SELECT * FROM prompt_tags ORDER BY prompt_id, tag_id"), tags
)
class TestApply(PairMetadataCase):
def snapshot_pairs(self):
return (
self.sql("SELECT * FROM prompt_pairs ORDER BY id"),
self.sql("SELECT * FROM prompt_pair_tags ORDER BY pair_id, tag_id"),
)
def test_apply_fills_only_empty_fields_and_leaves_pairs_alone(self):
_a, prompt = self.pair("apply me", "neg", category="portrait", tags=["a", "b"])
_b, partial = self.pair("partial", "n1", category="portrait", tags=["a"])
self.pair("partial", "n2", category="portrait", tags=["c"])
neg = self.sql("SELECT id FROM prompts WHERE text='neg'")[0][0]
pairs = self.snapshot_pairs()
stamps = self.sql("SELECT id, created_at, updated_at FROM prompts ORDER BY id")
state = self.sql("SELECT * FROM prompt_library_state ORDER BY prompt_id")
tag_rows = self.sql("SELECT id, name FROM tags ORDER BY id")
draft = self.preview()
result = self.apply(draft)
self.assertEqual(result["changed"], 2)
self.assertEqual(
result["rows"],
[
{"prompt_id": prompt, "category": "portrait", "tags": ["a", "b"]},
{"prompt_id": partial, "category": "portrait"},
],
)
self.assertEqual(result["categories"], 2)
self.assertEqual(result["tag_sets"], 1)
self.assertEqual(
(self.category(prompt), self.tags(prompt)), ("portrait", ["a", "b"])
)
self.assertEqual((self.category(partial), self.tags(partial)), ("portrait", []))
self.assertEqual((self.category(neg), self.tags(neg)), (None, []))
self.assertEqual(self.snapshot_pairs(), pairs)
self.assertEqual(
self.sql("SELECT id, created_at, updated_at FROM prompts ORDER BY id"),
stamps,
)
self.assertEqual(
self.sql("SELECT * FROM prompt_library_state ORDER BY prompt_id"), state
)
self.assertEqual(self.sql("SELECT id, name FROM tags ORDER BY id"), tag_rows)
def test_apply_returns_one_kept_row_per_protected_prompt(self):
_a, conflict = self.pair("conflict", "n1", category="asian")
self.pair("conflict", "n2", category="Adult")
_b, edited = self.pair("edited", category="portrait")
self.edit_stamp(edited)
_c, both = self.pair(
"both", "n1", category="x", tags=[f"t{n}" for n in range(21)]
)
self.pair("both", "n2", category="y", tags=[f"t{n}" for n in range(21)])
_d, filled = self.pair("fill", category="portrait")
draft = self.preview()
result = self.apply(draft)
self.assertEqual(
result["kept"],
[
{
"prompt_id": conflict,
"reasons": ["pairs disagree on category: Adult, asian"],
},
{
"prompt_id": edited,
"reasons": [
"edited after it was saved; its empty category or tags "
"may be intentional"
],
},
{
"prompt_id": both,
"reasons": [
"pairs disagree on category: x, y",
"its pair has more than 20 tags",
],
},
],
)
self.assertEqual(len(result["kept"]), draft.protected["count"])
self.assertEqual(self.category(filled), "portrait")
def test_kept_rows_are_never_capped(self):
for n in range(205):
_p, prompt = self.pair(f"kept {n:03d}", category="portrait")
self.edit_stamp(prompt)
result = self.apply(self.preview())
self.assertEqual(result["changed"], 0)
self.assertEqual(len(result["kept"]), 205)
def test_second_run_finds_nothing_and_a_partial_fill_completes_later(self):
_a, prompt = self.pair("partial", "n1", category="portrait", tags=["a"])
p2, _same = self.pair("partial", "n2", category="portrait", tags=["c"])
self.apply(self.preview())
again = self.preview()
self.assertEqual(again.scope["count"], 0)
self.assertEqual(again.protected["count"], 1) # tags still disagree
self.db.update_prompt_pair_metadata(p2, tags=["a"])
self.apply(self.preview())
self.assertEqual(self.tags(prompt), ["a"])
self.assertEqual(self.preview().protected["count"], 0)
def test_drift_raises_plan_changed_and_writes_nothing(self):
def pair_category(ids):
self.db.update_prompt_pair_metadata(ids[0], category="landscape")
def prompt_category(ids):
self.run_sql("UPDATE prompts SET category='mine' WHERE id=?", ids[1])
def new_pair(ids):
self.db.save_prompt_pair("drift", "other neg", category="landscape")
def tag_rename(ids):
tag = self.sql("SELECT id FROM tags WHERE name='a'")[0][0]
self.run_sql("UPDATE tags SET name='renamed' WHERE id=?", tag)
def edited(ids):
self.edit_stamp(ids[1])
for change in (pair_category, prompt_category, new_pair, tag_rename, edited):
with self.subTest(change=change.__name__):
ids = self.pair("drift", "n", category="portrait", tags=["a"])
draft = self.preview()
change(ids)
before = self.sql("SELECT id, category FROM prompts ORDER BY id")
tags = self.sql("SELECT * FROM prompt_tags ORDER BY prompt_id")
with self.assertRaises(maintenance.PlanChanged):
self.apply(draft)
self.assertEqual(
self.sql("SELECT id, category FROM prompts ORDER BY id"), before
)
self.assertEqual(
self.sql("SELECT * FROM prompt_tags ORDER BY prompt_id"), tags
)
self.run_sql("DELETE FROM prompt_pairs")
self.run_sql("DELETE FROM prompts")
self.run_sql("DELETE FROM tags")
def test_pair_run_between_preview_and_apply_still_applies(self):
pair_id, prompt = self.pair("busy", category="portrait")
draft = self.preview()
self.db.record_prompt_pair_use(pair_id)
self.apply(draft)
self.assertEqual(self.category(prompt), "portrait")
def test_guard_failure_rolls_back_everything(self):
_a, first = self.pair("first", category="portrait")
_b, second = self.pair("second", category="portrait")
draft = self.preview()
real = repair._fill_categories
def racing(conn, fills):
# Another writer's value appears inside the transaction.
conn.execute("UPDATE prompts SET category='x' WHERE id=?", (second,))
return real(conn, fills)
with patch.object(repair, "_fill_categories", side_effect=racing):
with self.assertRaises(maintenance.PlanChanged):
self.apply(draft)
self.assertIsNone(self.category(first))
self.assertIsNone(self.category(second))
def test_more_than_two_hundred_fills_are_all_returned(self):
for n in range(205):
self.pair(f"bulk {n:03d}", category="portrait")
result = self.apply(self.preview())
self.assertEqual(result["changed"], 205)
self.assertEqual(len(result["rows"]), 205)
def test_batches_bind_few_ids(self):
for n in range(5):
self.pair(f"batch {n}", category="portrait", tags=["a"])
with patch.object(maintenance, "ID_BATCH", 2):
draft = self.preview()
result = self.apply(draft)
self.assertEqual(result["changed"], 5)
if __name__ == "__main__":
unittest.main()
+82
View File
@@ -0,0 +1,82 @@
"""``release_free_memory``: glibc's malloc_trim(0), a no-op elsewhere (#523)."""
import sys
import unittest
from unittest import mock
from utils import native_memory
class ReleaseFreeMemoryTests(unittest.TestCase):
def setUp(self):
# Each test looks the symbol up again.
patcher = mock.patch.multiple(native_memory, _trim=None, _looked_up=False)
patcher.start()
self.addCleanup(patcher.stop)
@unittest.skipUnless(native_memory._glibc(), "glibc only")
def test_glibc_trims_once_per_call(self):
self.assertTrue(native_memory.release_free_memory())
self.assertTrue(native_memory.release_free_memory())
def test_other_platforms_do_nothing(self):
for platform in ("win32", "darwin", "cygwin"):
with self.subTest(platform), mock.patch.object(sys, "platform", platform):
native_memory._looked_up = False
with mock.patch.object(native_memory.ctypes, "CDLL") as cdll:
self.assertFalse(native_memory.release_free_memory())
cdll.assert_not_called()
def test_musl_does_nothing(self):
with (
mock.patch.object(sys, "platform", "linux"),
mock.patch.object(
native_memory.os, "confstr", side_effect=ValueError("unknown name")
),
mock.patch.object(native_memory.ctypes, "CDLL") as cdll,
):
self.assertFalse(native_memory.release_free_memory())
cdll.assert_not_called()
def test_a_missing_symbol_is_false(self):
library = mock.Mock(spec=[]) # no malloc_trim attribute
with (
mock.patch.object(native_memory, "_glibc", return_value=True),
mock.patch.object(native_memory.ctypes, "CDLL", return_value=library),
):
self.assertFalse(native_memory.release_free_memory())
def test_an_unloadable_libc_is_false(self):
with (
mock.patch.object(native_memory, "_glibc", return_value=True),
mock.patch.object(native_memory.ctypes, "CDLL", side_effect=OSError),
):
self.assertFalse(native_memory.release_free_memory())
def test_a_failing_trim_never_raises(self):
trim = mock.Mock(side_effect=OSError("boom"))
library = mock.Mock(malloc_trim=trim)
with (
mock.patch.object(native_memory, "_glibc", return_value=True),
mock.patch.object(native_memory.ctypes, "CDLL", return_value=library),
):
self.assertFalse(native_memory.release_free_memory())
trim.assert_called_once_with(0)
def test_the_symbol_is_looked_up_once(self):
trim = mock.Mock(return_value=1)
library = mock.Mock(malloc_trim=trim)
with (
mock.patch.object(native_memory, "_glibc", return_value=True),
mock.patch.object(
native_memory.ctypes, "CDLL", return_value=library
) as cdll,
):
self.assertTrue(native_memory.release_free_memory())
self.assertTrue(native_memory.release_free_memory())
cdll.assert_called_once_with(None)
self.assertEqual(trim.call_count, 2)
if __name__ == "__main__":
unittest.main()
+455
View File
@@ -0,0 +1,455 @@
"""PNG text without a pixel decode (#523), against the legacy reader floor.
Pillow's ``PngImageFile.text`` loads (decodes) every pixel to reach text
chunks written after IDAT. ``utils.png_text.png_text`` walks the remaining
chunks instead. The legacy reader (``Image.open(path).text``, every failure
read as ``{}``) is the floor: a file it read must read the same, and the
curated corpus below must read identically. The documented difference is
pixel data, which the walk never decodes: undecodable or cut-short pixel data
made the legacy reader fail, the walk still reads the text before it.
"""
import io
import random
import struct
import tempfile
import unittest
import zlib
from pathlib import Path
from unittest.mock import patch
from PIL import Image, ImageFile, PngImagePlugin
from utils.metadata_extractor import ComfyUIMetadataExtractor
from utils.png_text import png_text
FIXTURES = Path(__file__).resolve().parent / "fixtures"
IMAGE_SUFFIXES = {".png", ".jpg", ".jpeg", ".webp", ".gif", ".bmp"}
SIGNATURE = b"\x89PNG\r\n\x1a\n"
FUZZ_SEED = 523
FUZZ_CASES = 400
def legacy_text(path):
"""The legacy scan reader: Pillow's ``.text`` (a full decode)."""
try:
with Image.open(path) as image:
return dict(getattr(image, "text", None) or {})
except Exception:
return {}
def walked_text(path):
"""The new reader under the same failure rule."""
try:
with Image.open(path) as image:
return png_text(image)
except Exception:
return {}
def chunk(kind, data):
return (
struct.pack(">I", len(data))
+ kind
+ data
+ struct.pack(">I", zlib.crc32(kind + data) & 0xFFFFFFFF)
)
def chunks_of(raw):
"""``[(kind, data)]`` of a well-formed PNG."""
found, offset = [], 8
while offset < len(raw):
length, kind = struct.unpack(">I4s", raw[offset : offset + 8])
found.append((kind, raw[offset + 8 : offset + 8 + length]))
offset += 12 + length
return found
def assemble(parts):
return SIGNATURE + b"".join(
part if isinstance(part, bytes) else chunk(*part) for part in parts
)
def base_png(mode="RGB", size=(64, 64), text=None, noise=False):
"""A PNG from Pillow (text before IDAT), as ``[(kind, data)]``."""
image = Image.new(mode, size)
if noise:
rng = random.Random(7)
image = Image.frombytes(
"RGB", size, bytes(rng.randrange(256) for _ in range(size[0] * size[1] * 3))
)
info = PngImagePlugin.PngInfo()
for key, value in (text or {}).items():
info.add_text(key, value)
buffer = io.BytesIO()
image.save(buffer, format="PNG", pnginfo=info)
return chunks_of(buffer.getvalue())
def split_idat(parts, pieces=3):
"""Spread the IDAT payload over ``pieces`` consecutive IDAT chunks."""
data = b"".join(d for k, d in parts if k == b"IDAT")
step = max(1, len(data) // pieces)
idats = [(b"IDAT", data[i : i + step]) for i in range(0, len(data), step)]
first = next(i for i, (k, _) in enumerate(parts) if k == b"IDAT")
rest = [p for p in parts if p[0] != b"IDAT"]
return rest[:first] + idats + rest[first:]
def before_iend(parts, *extra):
return parts[:-1] + list(extra) + parts[-1:]
def text_chunk(key, value):
return (b"tEXt", key.encode("latin-1") + b"\0" + value.encode("latin-1"))
def ztxt_chunk(key, value):
return (b"zTXt", key.encode() + b"\0\0" + zlib.compress(value.encode("latin-1")))
def itxt_chunk(key, value, compressed=False):
body = value.encode("utf-8")
if compressed:
body = zlib.compress(body)
flag = b"\1\0" if compressed else b"\0\0"
return (b"iTXt", key.encode() + b"\0" + flag + b"en\0" + b"Title\0" + body)
def bad_crc(kind, data):
good = chunk(kind, data)
return good[:-4] + bytes(b ^ 0xFF for b in good[-4:])
PROMPT = '{"3": {"class_type": "KSampler", "inputs": {"seed": 1}}}'
WORKFLOW = '{"nodes": [], "links": []}'
def curated():
"""Name to PNG bytes; each must read the same with both readers."""
plain = base_png(text={"prompt": PROMPT})
trailing = before_iend(
base_png(), text_chunk("prompt", PROMPT), text_chunk("workflow", WORKFLOW)
)
cases = {
"no-text": assemble(base_png()),
"text-before-pixels": assemble(plain),
"text-after-pixels": assemble(trailing),
"ztxt-after-pixels": assemble(
before_iend(base_png(), ztxt_chunk("prompt", PROMPT))
),
"itxt-plain-and-compressed": assemble(
before_iend(
base_png(),
itxt_chunk("prompt", "café " + PROMPT),
itxt_chunk("workflow", WORKFLOW, compressed=True),
)
),
"xmp-after-pixels": assemble(
before_iend(
base_png(),
itxt_chunk("XML:com.adobe.xmp", "<x:xmpmeta/>"),
text_chunk("prompt", PROMPT),
)
),
"itxt-bad-utf8": assemble(
before_iend(
base_png(),
(b"iTXt", b"prompt\0\0\0\0\0\xff\xfe"),
text_chunk("workflow", WORKFLOW),
)
),
"later-copy-wins": assemble(
before_iend(plain, text_chunk("prompt", '{"later": 1}'))
),
"empty-key-and-no-separator": assemble(
before_iend(base_png(), (b"tEXt", b"\0value"), (b"tEXt", b"lonely"))
),
"split-pixels-text-between": assemble(
split_idat(before_iend(base_png(noise=True), text_chunk("prompt", PROMPT)))
),
"palette": assemble(
before_iend(base_png(mode="P"), text_chunk("prompt", PROMPT))
),
"sixteen-bit": assemble(
before_iend(base_png(mode="I;16"), text_chunk("prompt", PROMPT))
),
"bad-crc-before-pixels": assemble(
[base_png()[0], bad_crc(*text_chunk("prompt", PROMPT))] + base_png()[1:]
),
"bad-crc-after-pixels": assemble(
base_png()[:-1] + [bad_crc(*text_chunk("prompt", PROMPT))] + base_png()[-1:]
),
"unknown-chunk-after-pixels": assemble(
before_iend(base_png(), (b"prVt", b"x" * 40), text_chunk("prompt", PROMPT))
),
"garbage-chunk-id-after-pixels": assemble(
base_png()[:-1]
+ [b"\0\0\0\4\x00\x01\x02\x03abcd\0\0\0\0"]
+ [chunk(*text_chunk("prompt", PROMPT))]
+ base_png()[-1:]
),
"oversized-ztxt-after-pixels": assemble(
before_iend(
base_png(),
(
b"zTXt",
b"prompt\0\0"
+ zlib.compress(b"a" * (PngImagePlugin.MAX_TEXT_CHUNK + 10)),
),
)
),
"pixels-length-past-end": assemble(
[p for p in base_png() if p[0] != b"IDAT" and p[0] != b"IEND"]
)
+ b"\xff\xff\x00\x00IDAT"
+ b"\0" * 64,
}
full = assemble(trailing)
idat_at = full.index(b"IDAT")
cases["truncated-in-pixels"] = full[: idat_at + 20]
cases["no-end-after-pixels"] = full[: full.rindex(b"IEND") - 4]
cases["truncated-in-trailing-text"] = full[: full.rindex(b"workflow") + 4]
cases["unknown-chunk-past-end"] = (
assemble(before_iend(base_png(), ztxt_chunk("prompt", PROMPT))[:-1])
+ chunk(b"prVt", b"x" * 40)[:-20]
)
cases["not-a-png"] = b"nope, not an image at all"
cases["signature-only"] = SIGNATURE
return cases
def corrupt_pixels():
"""Complete IDAT chunks with valid CRCs over data zlib cannot inflate."""
parts = before_iend(base_png(), text_chunk("prompt", PROMPT))
return assemble(
[(k, b"\x78\x9c" + b"\xff" * 30) if k == b"IDAT" else (k, d) for k, d in parts]
)
def apng():
frames = [Image.new("RGB", (16, 16), colour) for colour in ("red", "blue")]
info = PngImagePlugin.PngInfo()
info.add_text("prompt", PROMPT)
buffer = io.BytesIO()
frames[0].save(
buffer, format="PNG", save_all=True, append_images=frames[1:], pnginfo=info
)
return assemble(
before_iend(chunks_of(buffer.getvalue()), text_chunk("workflow", WORKFLOW))
)
def fixture_images():
return sorted(p for p in FIXTURES.rglob("*") if p.suffix.lower() in IMAGE_SUFFIXES)
class _TempFiles(unittest.TestCase):
def setUp(self):
self._tmp = tempfile.TemporaryDirectory()
self.addCleanup(self._tmp.cleanup)
def write(self, name, raw):
path = Path(self._tmp.name) / name
path.write_bytes(raw)
return path
class PngTextParityTests(_TempFiles):
def test_every_fixture_image_reads_like_the_legacy_reader(self):
images = fixture_images()
self.assertGreaterEqual(len(images), 25)
for path in images:
with self.subTest(path=path.relative_to(FIXTURES)):
self.assertEqual(walked_text(path), legacy_text(path))
def test_curated_corpus_reads_like_the_legacy_reader(self):
for name, raw in curated().items():
with self.subTest(name):
path = self.write(name + ".png", raw)
self.assertEqual(walked_text(path), legacy_text(path))
def test_curated_corpus_exercises_trailing_text(self):
"""The corpus is not vacuous: trailing text exists and is read."""
cases = curated()
path = self.write("t.png", cases["text-after-pixels"])
self.assertEqual(walked_text(path), {"prompt": PROMPT, "workflow": WORKFLOW})
with Image.open(path) as image:
self.assertNotIn("prompt", image.info)
self.assertEqual(
walked_text(self.write("x.png", cases["truncated-in-pixels"])), {}
)
def test_itxt_values_keep_their_language_and_type(self):
path = self.write("i.png", curated()["itxt-plain-and-compressed"])
new, old = walked_text(path), legacy_text(path)
self.assertIsInstance(new["prompt"], PngImagePlugin.iTXt)
self.assertEqual(new["prompt"].lang, old["prompt"].lang)
self.assertEqual(new["prompt"].tkey, old["prompt"].tkey)
def test_animated_png_falls_back_to_the_legacy_read(self):
path = self.write("a.png", apng())
self.assertEqual(walked_text(path), legacy_text(path))
self.assertEqual(walked_text(path)["workflow"], WORKFLOW)
def test_truncated_pixels_read_alike_when_truncated_images_load(self):
"""With LOAD_TRUNCATED_IMAGES set (a node pack may set it) both readers
accept a short file alike."""
path = self.write("t.png", curated()["truncated-in-pixels"])
with patch.object(ImageFile, "LOAD_TRUNCATED_IMAGES", True):
self.assertEqual(walked_text(path), legacy_text(path))
def test_a_text_decode_error_ends_the_walk_like_legacy(self):
"""Pillow's ``load_end`` stops at a chunk whose handler raises
``UnicodeDecodeError`` and keeps the text read so far."""
path = self.write(
"u.png",
assemble(
before_iend(
base_png(),
ztxt_chunk("prompt", PROMPT),
text_chunk("workflow", WORKFLOW),
ztxt_chunk("parameters", "steps 20"),
)
),
)
def undecodable(stream, position, length):
raise UnicodeDecodeError("latin-1", b"", 0, 1, "test")
with patch.object(PngImagePlugin.PngStream, "chunk_tEXt", undecodable):
old = legacy_text(path)
self.assertEqual(walked_text(path), old)
self.assertEqual(old, {"prompt": PROMPT})
def test_the_text_memory_budget_counts_text_before_the_pixels(self):
"""Pillow's MAX_TEXT_MEMORY spans the whole file: text before and
after the pixels, each under it, fail together in both readers."""
path = self.write(
"m.png",
assemble(
before_iend(
base_png(text={"prompt": PROMPT}), text_chunk("workflow", WORKFLOW)
)
),
)
budget = len(PROMPT) + len(WORKFLOW) - 1
with patch.object(PngImagePlugin, "MAX_TEXT_MEMORY", budget):
self.assertEqual(legacy_text(path), {})
self.assertEqual(walked_text(path), {})
with patch.object(PngImagePlugin, "MAX_TEXT_MEMORY", budget + 1):
self.assertEqual(walked_text(path), legacy_text(path))
self.assertEqual(len(walked_text(path)), 2)
def test_complete_but_undecodable_pixels_read_more_than_legacy(self):
path = self.write("c.png", corrupt_pixels())
self.assertEqual(legacy_text(path), {})
self.assertEqual(walked_text(path), {"prompt": PROMPT})
def test_cut_short_pixels_read_the_text_before_them(self):
full = assemble(base_png(text={"prompt": PROMPT}))
path = self.write("s.png", full[: full.index(b"IDAT") + 12])
self.assertEqual(legacy_text(path), {})
self.assertEqual(walked_text(path), {"prompt": PROMPT})
def test_mutated_files_never_read_less_than_legacy(self):
"""Seeded fuzz: truncations and byte flips of a file with text before,
between and after its pixel chunks. Wherever the legacy reader found
text, the walk finds exactly the same."""
base = assemble(
split_idat(
before_iend(
base_png(text={"workflow": WORKFLOW}),
ztxt_chunk("prompt", PROMPT),
itxt_chunk("parameters", "steps 20"),
)
)
)
rng = random.Random(FUZZ_SEED)
compared = 0
for case in range(FUZZ_CASES):
raw = bytearray(base)
if case % 2:
del raw[rng.randrange(8, len(raw)) :]
else:
for _ in range(rng.randrange(1, 4)):
raw[rng.randrange(8, len(raw))] = rng.randrange(256)
path = self.write(f"fuzz{case}.png", bytes(raw))
old = legacy_text(path)
if old:
compared += 1
with self.subTest(case=case):
self.assertEqual(walked_text(path), old)
self.assertGreater(compared, FUZZ_CASES // 5)
def test_non_png_images_have_no_text(self):
for name in ("tiny.jpg", "tiny.webp"):
with Image.open(FIXTURES / "local-inspect" / name) as image:
self.assertEqual(png_text(image), {})
class NoPixelDecodeTests(_TempFiles):
"""Metadata reads never decode pixels (the #523 memory finding)."""
def png_fixtures(self):
return sorted((FIXTURES / "inspection_parity").glob("*.png"))
def test_png_text_does_not_load_pixels_and_leaves_the_image_usable(self):
path = self.write("t.png", curated()["text-after-pixels"])
with Image.open(path) as image:
with patch.object(ImageFile.ImageFile, "load", autospec=True) as load:
self.assertEqual(png_text(image)["workflow"], WORKFLOW)
load.assert_not_called()
image.load()
self.assertEqual(image.size, (64, 64))
self.assertEqual(image.text["workflow"], WORKFLOW)
def test_scan_reader_does_not_decode_pixels(self):
from py.api import PromptManagerAPI
api = PromptManagerAPI.__new__(PromptManagerAPI)
fixtures = self.png_fixtures() + [
self.write("t.png", curated()["text-after-pixels"])
]
with patch.object(ImageFile.ImageFile, "load", autospec=True) as load:
found = [api._extract_comfyui_metadata(str(p)) for p in fixtures]
load.assert_not_called()
self.assertEqual(found, [legacy_text(p) for p in fixtures])
self.assertTrue(all(found))
def test_monitor_extractor_does_not_decode_pixels(self):
extractor = ComfyUIMetadataExtractor()
fixtures = self.png_fixtures()
with patch.object(ImageFile.ImageFile, "load", autospec=True) as load:
found = [extractor.extract_metadata(str(p)) for p in fixtures]
load.assert_not_called()
self.assertTrue(all(metadata and len(metadata) > 1 for metadata in found))
class MonitorExtractorParityTests(_TempFiles):
"""``extract_metadata`` answers exactly as with the legacy ``.text``."""
def legacy_extract(self, path):
def old_text(image):
return dict(getattr(image, "text", None) or {})
with patch("utils.metadata_extractor.png_text", old_text):
return ComfyUIMetadataExtractor().extract_metadata(str(path))
def test_every_fixture_and_curated_file_extracts_like_legacy(self):
paths = fixture_images() + [
self.write(name + ".png", raw) for name, raw in curated().items()
]
extractor = ComfyUIMetadataExtractor()
for path in paths:
with self.subTest(path=path.name):
self.assertEqual(
extractor.extract_metadata(str(path)), self.legacy_extract(path)
)
if __name__ == "__main__":
unittest.main()
+4 -3
View File
@@ -291,7 +291,7 @@ class TestThumbnailOnFirstView(AutoThumbnailCase):
await self._writers(state, auto.ON_DEMAND_LIMIT)
await self._waiting(len(names))
self.assertEqual(default_calls, [], "no view waits on a default thread")
self.assertTrue(all(name.startswith("pm-preview") for name in state["threads"]))
self.assertTrue(all(name.startswith("pm-image") for name in state["threads"]))
status = await self.client.get("/prompt_manager/images/thumbnails/status")
self.assertEqual(status.status, 200, "another route answers meanwhile")
await status.read()
@@ -678,7 +678,7 @@ class TestShutdown(AutoThumbnailCase):
_entry(self.api, self.output_dir, "viewed.png")["thumbnail_url"]
)
await resp.read()
pool_threads = list(self.api._preview_pool_threads())
pool_threads = list(self.api._image_pool_threads())
self.assertTrue(pool_threads)
loop = asyncio.get_running_loop()
await loop.run_in_executor(
@@ -904,7 +904,8 @@ class TestThumbnailOnCapture(AutoThumbnailCase):
self.api.queue_capture_thumbnail(str(path))
release.set()
await self._idle()
self.assertEqual(batches, [(1, 1), (2, 2), (2, 2)])
# The job pool's size is the setting; each batch is at most that (#523).
self.assertEqual(batches, [(1, 2), (2, 2), (2, 2)])
self.assertLessEqual(state["peak"], 2)
for i in range(5):
self.assertTrue(
+110 -4
View File
@@ -1,4 +1,8 @@
"""Tests for utils/parallel.py: bounded thread-pool map used by long jobs."""
"""Tests for utils/parallel.py: bounded thread-pool map used by long jobs.
The jobs share one long-lived pool of Worker threads (#523): its threads (and
their malloc arenas) are reused from job to job instead of a new set per call.
"""
import os
import sys
@@ -8,10 +12,22 @@ import unittest
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from utils.parallel import map_parallel # noqa: E402
from utils import parallel # noqa: E402
from utils.parallel import JOB_THREAD_PREFIX, map_parallel # noqa: E402
WAIT = 30
def _job_threads():
return {
t for t in threading.enumerate() if t.name.startswith(JOB_THREAD_PREFIX + "_")
}
class TestMapParallel(unittest.TestCase):
def setUp(self):
self.assertTrue(parallel.shutdown_job_pool(WAIT))
self.addCleanup(parallel.shutdown_job_pool, WAIT)
def test_results_keep_input_order(self):
def slow_square(n):
@@ -22,12 +38,102 @@ class TestMapParallel(unittest.TestCase):
map_parallel(slow_square, [1, 2, 3, 4], workers=4), [1, 4, 9, 16]
)
def test_one_worker_runs_on_the_calling_thread(self):
def test_one_worker_runs_on_the_one_job_thread(self):
names = []
map_parallel(
lambda _: names.append(threading.current_thread().name), [1, 2, 3], 1
)
self.assertEqual(set(names), {threading.current_thread().name})
self.assertEqual(len(set(names)), 1)
self.assertTrue(names[0].startswith(JOB_THREAD_PREFIX + "_"), names)
def test_jobs_reuse_the_same_long_lived_threads(self):
"""Arenas are reused instead of a new set of threads per job."""
first, second = set(), set()
map_parallel(lambda _: first.add(threading.current_thread()), range(8), 2)
map_parallel(lambda _: second.add(threading.current_thread()), range(8), 2)
self.assertTrue(second <= first, (first, second))
self.assertLessEqual(len(_job_threads()), 2)
self.assertTrue(all(t.is_alive() for t in first | second))
def test_a_new_worker_count_replaces_the_pool(self):
map_parallel(lambda n: n, range(4), 2)
old = _job_threads()
map_parallel(lambda n: n, range(4), 3)
for thread in old:
thread.join(WAIT)
self.assertFalse(thread.is_alive(), thread.name)
self.assertLessEqual(len(_job_threads()), 3)
def test_shutdown_also_joins_a_pool_a_resize_retired(self):
entered, release = threading.Event(), threading.Event()
def hold(_):
entered.set()
release.wait(WAIT)
caller = threading.Thread(target=map_parallel, args=(hold, [1], 1))
caller.start()
self.addCleanup(caller.join, WAIT)
self.addCleanup(release.set)
self.assertTrue(entered.wait(WAIT))
map_parallel(lambda n: n, [1, 2], 2) # a new size retires the busy pool
self.assertFalse(parallel.shutdown_job_pool(0.05))
release.set()
caller.join(WAIT)
self.assertTrue(parallel.shutdown_job_pool(WAIT))
self.assertFalse(parallel.job_pool_started())
def test_an_item_run_from_a_job_thread_runs_inline(self):
"""A nested map never waits for the threads it is running on."""
def outer(n):
return map_parallel(lambda m: (m, threading.current_thread()), [n], 1)
results = map_parallel(outer, [1, 2], 1)
self.assertEqual([r[0][0] for r in results], [1, 2])
def test_shutdown_stops_the_threads_and_a_later_job_starts_fresh(self):
map_parallel(lambda n: n, range(4), 2)
threads = _job_threads()
self.assertTrue(threads)
self.assertTrue(parallel.job_pool_started())
self.assertTrue(parallel.shutdown_job_pool(WAIT))
self.assertFalse(parallel.job_pool_started())
self.assertFalse(any(t.is_alive() for t in threads))
self.assertEqual(map_parallel(lambda n: n + 1, [1], 2), [2])
def test_shutdown_reports_a_thread_still_running(self):
entered, release = threading.Event(), threading.Event()
def hold(_):
entered.set()
release.wait(WAIT)
caller = threading.Thread(target=map_parallel, args=(hold, [1], 1))
caller.start()
self.assertTrue(entered.wait(WAIT))
self.assertFalse(parallel.shutdown_job_pool(0.05))
release.set()
caller.join(WAIT)
self.assertFalse(caller.is_alive())
def test_a_failing_item_waits_for_the_running_ones(self):
"""The caller sees the error only after its other items stopped."""
running = threading.Event()
finished = []
def item(n):
if n == 0:
self.assertTrue(running.wait(WAIT))
raise ValueError("zero")
running.set()
time.sleep(0.05)
finished.append(n)
return n
with self.assertRaises(ValueError):
map_parallel(item, [0, 1], 2)
self.assertEqual(finished, [1])
def test_several_workers_use_several_threads(self):
names = set()
+395
View File
@@ -0,0 +1,395 @@
"""Measure the server's resident memory under a busy Images/Library session (#523).
Manual only, Linux only (it reads ``/proc`` and glibc's ``malloc_info``); never
run in CI. It starts the production API inside the disposable browser test
server (``tests/browser/server.py``: its own temporary library, database and
output folder; never a live ComfyUI) and drives it for ``--minutes``:
- ``--clients`` threads loop over: a workspace grid page of 100 images, 40 of
its thumbnails, 6 inspector reads and 6 full-file reads, the Library list
with 3 entries' history and images, and the recent prompt and image lists;
- every minute, ``--batch`` new PNGs (832 to 1536 px, ComfyUI-shaped
metadata) are copied into the output folder, then a scan job and a
thumbnail job run to completion.
Every 5 s the server samples its RSS, Python heap (tracemalloc), threads and
glibc arena count. The test server pins glibc's mmap threshold (#516); this
driver turns that off so the production allocator behaviour is measured,
unless ``--mallopt`` asks for the pinned comparison run.
python tools/perf/memory_drive.py --label after --minutes 12
python tools/perf/memory_drive.py --repo /path/to/other/checkout --label before
The checkout under ``--repo`` (default: this one) is the code measured; the
driver itself only needs Pillow. The last line printed is a Markdown table row.
"""
import argparse
import ctypes
import json
import os
import random
import shutil
import subprocess # nosec B404 - starts this script's own server child
import sys
import tempfile
import threading
import time
import tracemalloc
import urllib.parse
import urllib.request
from concurrent.futures import ThreadPoolExecutor
HERE = os.path.dirname(os.path.abspath(__file__))
REPO = os.path.dirname(os.path.dirname(HERE))
PORT_ANNOUNCEMENT = "PM_TEST_SERVER_PORT="
OUTPUT_ANNOUNCEMENT = "output="
SAMPLE_SECONDS = 5
MEDIA_FILES = 120
MEDIA_SIDES = (832, 1024, 1216, 1536)
REQUEST_TIMEOUT = 120
JOB_POLL_LIMIT = 600
FINAL_STATES = ("finished", "cancelled", "failed")
M_MMAP_THRESHOLD = -3
# ── the server child ────────────────────────────────────────────────────
def _arena_count(libc):
"""glibc's arena count (``<heap nr=`` entries of ``malloc_info``)."""
buffer, size = ctypes.c_char_p(), ctypes.c_size_t()
stream = libc.open_memstream(ctypes.byref(buffer), ctypes.byref(size))
libc.malloc_info(0, stream)
libc.fclose(stream)
text = ctypes.string_at(buffer, size.value).decode()
libc.free(ctypes.cast(buffer, ctypes.c_void_p))
return text.count("<heap nr=")
def _status():
with open("/proc/self/status", encoding="ascii") as status:
return {
line.split(":")[0]: int(line.split()[1])
for line in status
if line.startswith(("VmRSS", "RssAnon", "Threads"))
}
def _sample_forever(libc, samples_path):
start = time.monotonic()
announced = False
with open(samples_path, "a", buffering=1, encoding="utf-8") as out:
while True:
folder_paths = sys.modules.get("folder_paths")
if not announced and hasattr(folder_paths, "get_output_directory"):
out.write(OUTPUT_ANNOUNCEMENT + folder_paths.get_output_directory())
out.write("\n")
announced = True
status = _status()
out.write(
json.dumps(
{
"t": round(time.monotonic() - start),
"rss": status["VmRSS"] >> 10,
"anon": status["RssAnon"] >> 10,
"heap": tracemalloc.get_traced_memory()[0] >> 20,
"threads": status["Threads"],
"arenas": _arena_count(libc),
}
)
+ "\n"
)
time.sleep(SAMPLE_SECONDS if announced else 0.5)
def serve(repo, samples_path, mallopt):
"""Run the browser test server of *repo* in this process, sampled."""
sys.path.insert(0, repo)
os.chdir(repo)
tracemalloc.start(1)
libc = ctypes.CDLL(None)
libc.open_memstream.restype = ctypes.c_void_p
libc.open_memstream.argtypes = [
ctypes.POINTER(ctypes.c_char_p),
ctypes.POINTER(ctypes.c_size_t),
]
libc.fclose.argtypes = [ctypes.c_void_p]
libc.malloc_info.argtypes = [ctypes.c_int, ctypes.c_void_p]
libc.free.argtypes = [ctypes.c_void_p]
if mallopt:
libc.mallopt(M_MMAP_THRESHOLD, 128 * 1024)
threading.Thread(
target=_sample_forever, args=(libc, samples_path), daemon=True
).start()
from tests.browser import server
if not mallopt:
server.pin_mmap_threshold = lambda: False # production allocator
server.main(["--port", "0", "--exit-on-stdin-eof"])
# ── the driver ──────────────────────────────────────────────────────────
def make_media(folder, count=MEDIA_FILES, seed=523):
"""Synthetic ComfyUI-shaped PNGs (smooth noise, prompt and workflow)."""
from PIL import Image, PngImagePlugin
rng = random.Random(seed)
os.makedirs(folder, exist_ok=True)
for index in range(count):
side = MEDIA_SIDES[index % len(MEDIA_SIDES)]
small = Image.frombytes(
"RGB", (64, 64), bytes(rng.randrange(256) for _ in range(64 * 64 * 3))
)
image = small.resize((side, side), Image.BICUBIC)
text = f"synthetic scene {index}, soft light, detailed"
prompt = {
"3": {
"class_type": "KSampler",
"inputs": {
"seed": index,
"steps": 20,
"cfg": 7.0,
"sampler_name": "euler",
"scheduler": "normal",
"denoise": 1.0,
"model": ["4", 0],
"positive": ["6", 0],
"negative": ["7", 0],
"latent_image": ["5", 0],
},
},
"4": {
"class_type": "CheckpointLoaderSimple",
"inputs": {"ckpt_name": "model.safetensors"},
},
"5": {
"class_type": "EmptyLatentImage",
"inputs": {"width": side, "height": side, "batch_size": 1},
},
"6": {
"class_type": "CLIPTextEncode",
"inputs": {"text": text, "clip": ["4", 1]},
},
"7": {
"class_type": "CLIPTextEncode",
"inputs": {"text": "blurry", "clip": ["4", 1]},
},
"9": {
"class_type": "SaveImage",
"inputs": {"filename_prefix": "img", "images": ["8", 0]},
},
}
workflow = {
"nodes": [
{"id": int(k), "type": v["class_type"]} for k, v in prompt.items()
]
}
info = PngImagePlugin.PngInfo()
info.add_text("prompt", json.dumps(prompt))
info.add_text("workflow", json.dumps(workflow))
image.save(
os.path.join(folder, f"m{index:03d}.png"), pnginfo=info, compress_level=4
)
class Driver:
def __init__(self, base, output, media, batch):
self.base, self.output, self.media, self.batch = base, output, media, batch
self.errors, self.batches = [], 0
self.names = sorted(os.listdir(media))
def call(self, method, path, body=None, raw=False):
data = None if body is None else json.dumps(body).encode()
request = urllib.request.Request( # nosec B310 - loopback test server
self.base + path,
data=data,
method=method,
headers={"Content-Type": "application/json"},
)
try:
with urllib.request.urlopen( # nosec B310
request, timeout=REQUEST_TIMEOUT
) as response:
payload = response.read()
except Exception as error: # keep driving; errors are counted
self.errors.append(f"{method} {path.split('?')[0]}: {error}")
return None
return payload if raw else json.loads(payload or b"{}")
def add_batch(self):
self.batches += 1
folder = os.path.join(self.output, f"batch{self.batches:03d}")
os.makedirs(folder)
for index in range(self.batch):
name = self.names[(self.batches * self.batch + index) % len(self.names)]
shutil.copyfile(
os.path.join(self.media, name),
os.path.join(folder, f"img_{self.batches:03d}_{index:03d}.png"),
)
def wait_job(self, response):
job = ((response or {}).get("job") or {}).get("id")
for _ in range(JOB_POLL_LIMIT if job else 0):
state = self.call("GET", f"/prompt_manager/jobs/{job}")
if not state or (state.get("job") or {}).get("state") in FINAL_STATES:
return
time.sleep(1)
def scan_and_thumbnails(self):
self.add_batch()
self.wait_job(self.call("POST", "/prompt_manager/jobs/scan", {}))
self.wait_job(self.call("POST", "/prompt_manager/jobs/thumbnails", {}))
def browse(self, _index=None):
offset = random.randint(0, 3) * 100 # nosec B311 - load shape only
page = self.call(
"GET",
f"/prompt_manager/images/output?mode=workspace&limit=100&offset={offset}",
)
images = [i for i in (page or {}).get("images") or [] if isinstance(i, dict)]
urls = [i.get("thumbnail_url") or i.get("url") for i in images]
urls = [u for u in urls if u]
for url in random.sample(urls, min(40, len(urls))):
self.call("GET", url, raw=True)
for image in random.sample(images, min(6, len(images))):
self.inspect(image)
if image.get("url"):
self.call("GET", image["url"], raw=True)
library = self.call("GET", "/prompt_manager/library?limit=100") or {}
entries = [e for e in library.get("items") or [] if isinstance(e, dict)]
for entry in random.sample(entries, min(3, len(entries))):
prompt_id = entry.get("id") or entry.get("positive_id")
if prompt_id is not None:
self.call(
"GET", f"/prompt_manager/library/{prompt_id}/history", raw=True
)
self.call(
"GET", f"/prompt_manager/library/{prompt_id}/images", raw=True
)
self.call("GET", "/prompt_manager/recent?limit=50", raw=True)
self.call("GET", "/prompt_manager/images/recent?limit=50", raw=True)
def inspect(self, image):
relative, index = image.get("path"), image.get("root_index")
if not relative or not isinstance(index, int):
return
query = {"root": str(index)}
if image.get("root_scope"):
query["root_scope"] = image["root_scope"]
self.call(
"GET",
"/prompt_manager/images/inspect/"
+ urllib.parse.quote(relative)
+ "?"
+ urllib.parse.urlencode(query),
raw=True,
)
def _wait_for_output(samples_path, process):
while process.poll() is None:
with open(samples_path, encoding="utf-8") as samples:
for line in samples:
if line.startswith(OUTPUT_ANNOUNCEMENT):
return line[len(OUTPUT_ANNOUNCEMENT) :].strip()
time.sleep(0.2)
raise RuntimeError("the server exited before it started")
def summarize(samples_path, label, batches, errors):
rows = []
with open(samples_path, encoding="utf-8") as samples:
for line in samples:
if line.startswith("{"):
rows.append(json.loads(line))
if not rows:
raise RuntimeError("no samples were taken")
rss = [row["rss"] for row in rows]
late = [row for row in rows if row["t"] >= 120] or rows
end = rows[-1]
return (
f"| {label} | {rss[0]} → {end['rss']} | {min(r['rss'] for r in late)}–"
f"{max(r['rss'] for r in late)} | {max(rss)} | {rows[0]['heap']} → "
f"{end['heap']} | {max(r['threads'] for r in rows)} | "
f"{max(r['arenas'] for r in rows)} | {batches} | {len(errors)} |"
)
def drive(args):
work = args.out or tempfile.mkdtemp(prefix="pm-memory-")
media = os.path.join(work, "media")
if not os.path.isdir(media):
print(f"Writing {MEDIA_FILES} PNGs to {media}", flush=True)
make_media(media)
samples_path = os.path.join(work, f"{args.label}.samples")
open(samples_path, "w", encoding="utf-8").close()
command = [sys.executable, os.path.abspath(__file__), "--serve", samples_path]
command += ["--repo", os.path.abspath(args.repo)]
command += ["--mallopt"] if args.mallopt else []
with open(os.path.join(work, f"{args.label}.stderr"), "w") as stderr:
process = subprocess.Popen( # nosec B603
command, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=stderr
)
try:
port = None
for line in process.stdout:
text = line.decode(errors="replace")
if text.startswith(PORT_ANNOUNCEMENT):
port = int(text[len(PORT_ANNOUNCEMENT) :])
break
if port is None:
raise RuntimeError("the server announced no port")
output = _wait_for_output(samples_path, process)
driver = Driver(f"http://127.0.0.1:{port}", output, media, args.batch)
run(driver, args)
finally:
process.stdin.close()
process.wait(60)
print(f"Samples: {samples_path}")
print(
"| run | RSS start → end (MB) | after 2 min | max | heap (MB) | threads "
"| arenas | batches | errors |"
)
print(summarize(samples_path, args.label, driver.batches, driver.errors))
def run(driver, args):
driver.call("POST", "/__test__/reset", {})
driver.scan_and_thumbnails()
deadline = time.monotonic() + args.minutes * 60
next_batch = time.monotonic() + 60
with ThreadPoolExecutor(args.clients) as pool:
while time.monotonic() < deadline:
list(pool.map(driver.browse, range(args.clients)))
if time.monotonic() > next_batch:
driver.scan_and_thumbnails()
next_batch = time.monotonic() + 60
time.sleep(SAMPLE_SECONDS + 1) # one more sample after the load stops
def main(argv=None):
parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
parser.add_argument("--repo", default=REPO, help="checkout to measure")
parser.add_argument("--label", default="run")
parser.add_argument("--minutes", type=float, default=12)
parser.add_argument("--clients", type=int, default=6)
parser.add_argument("--batch", type=int, default=60, help="new PNGs a minute")
parser.add_argument("--out", help="work folder (media, samples); default temp")
parser.add_argument(
"--mallopt", action="store_true", help="pin the mmap threshold (comparison)"
)
parser.add_argument("--serve", metavar="SAMPLES", help=argparse.SUPPRESS)
args = parser.parse_args(argv)
if not sys.platform.startswith("linux"):
parser.error("Linux only: it reads /proc and glibc's malloc_info")
if args.serve:
serve(os.path.abspath(args.repo), args.serve, args.mallopt)
else:
drive(args)
if __name__ == "__main__":
main()
+12 -9
View File
@@ -37,6 +37,7 @@ from typing import Optional, Dict, Any
from PIL import Image
from .logging_config import get_logger
from .png_text import png_text
class ComfyUIMetadataExtractor:
@@ -88,12 +89,14 @@ class ComfyUIMetadataExtractor:
# Add basic file information
metadata["file_info"] = self.get_file_info(image_path, image)
# Extract ComfyUI-specific metadata from PNG text chunks
if hasattr(image, "text") and image.text:
# Extract ComfyUI-specific metadata from PNG text chunks,
# read without decoding the pixels (#523)
text = png_text(image)
if text:
# Look for ComfyUI workflow data
if "workflow" in image.text:
if "workflow" in text:
try:
workflow_data = json.loads(image.text["workflow"])
workflow_data = json.loads(text["workflow"])
metadata["workflow"] = workflow_data
# Extract text encoder nodes from workflow
@@ -107,9 +110,9 @@ class ComfyUIMetadataExtractor:
self.logger.warning(f"Failed to parse workflow JSON: {e}")
# Look for prompt data
if "prompt" in image.text:
if "prompt" in text:
try:
prompt_data = json.loads(image.text["prompt"])
prompt_data = json.loads(text["prompt"])
metadata["prompt"] = prompt_data
except json.JSONDecodeError as e:
self.logger.warning(f"Failed to parse prompt JSON: {e}")
@@ -129,13 +132,13 @@ class ComfyUIMetadataExtractor:
]
for field in metadata_fields:
if field in image.text:
if field in text:
try:
# Try to parse as JSON first
metadata[field] = json.loads(image.text[field])
metadata[field] = json.loads(text[field])
except (json.JSONDecodeError, TypeError):
# Store as string if not valid JSON
metadata[field] = image.text[field]
metadata[field] = text[field]
return (
metadata
+65
View File
@@ -0,0 +1,65 @@
"""Give glibc's free heap pages back to the OS after a long image job (#523).
Image work (decode, resize, encode) leaves image-sized free blocks in the
malloc arenas of the threads that did it; glibc keeps those pages resident.
``release_free_memory`` calls ``malloc_trim(0)`` once, which returns the free
pages of every arena and changes no allocator setting (no ``mallopt``, no
``MALLOC_ARENA_MAX``: ComfyUI owns this process's allocator).
glibc only: on Windows, macOS and musl (Alpine) it does nothing. It never
raises; a missing or failing ``malloc_trim`` is a quiet ``False``.
"""
import ctypes
import os
import sys
import threading
_lock = threading.Lock()
_trim = None
_looked_up = False
def _glibc() -> bool:
"""Whether this process runs on glibc (musl has no CS_GNU_LIBC_VERSION)."""
if not sys.platform.startswith("linux"):
return False
try:
version = os.confstr("CS_GNU_LIBC_VERSION")
except (AttributeError, ValueError, OSError):
return False
return bool(version) and version.startswith("glibc")
def _malloc_trim():
"""glibc's ``malloc_trim``, looked up once; None where there is none."""
global _trim, _looked_up
with _lock:
if not _looked_up:
_looked_up = True
_trim = _lookup()
return _trim
def _lookup():
if not _glibc():
return None
try:
trim = ctypes.CDLL(None).malloc_trim
except (OSError, AttributeError):
return None
trim.argtypes = [ctypes.c_size_t]
trim.restype = ctypes.c_int
return trim
def release_free_memory() -> bool:
"""Return free heap pages to the OS once (glibc); True when it ran."""
trim = _malloc_trim()
if trim is None:
return False
try:
trim(0)
except Exception: # a foreign libc symbol: never fail the caller's job
return False
return True
+114 -22
View File
@@ -15,18 +15,36 @@ create new thread at interpreter shutdown") only once the interpreter is
marked finalizing, after the ``atexit`` callbacks. ``refused_at_shutdown``
recognises both; ``note_refusal`` logs one DEBUG line per place, with no
traceback.
One long-lived pool (#523): every ``map_parallel`` call shares one pool of
``workers`` threads (``JOB_THREAD_PREFIX``), kept between jobs, instead of a
new pool per call. Image work then reuses the same few threads, and their
malloc arenas, rather than leaving each call's image-sized buffers in a new
set of arenas. A different ``workers`` count (the Worker threads setting
changed) replaces the pool; the old one finishes what it holds and exits.
``shutdown_job_pool`` stops it (the package teardown); the next call starts
a fresh one.
"""
import logging
import threading
from concurrent.futures import ThreadPoolExecutor
import time
from concurrent.futures import ThreadPoolExecutor, wait
from .exit_signal import checked
SHUTDOWN_REFUSAL = "interpreter shutdown"
# ``submit`` on a pool already shut down (this module's teardown or a resize).
SHUT_POOL_REFUSAL = "after shutdown"
JOB_THREAD_PREFIX = "pm-worker"
# The package teardown's bound for the job threads to finish their item.
JOB_POOL_JOIN_SECONDS = 5
_logger = logging.getLogger("prompt_manager.parallel")
_noted = set()
_noted_lock = threading.Lock()
_pool_lock = threading.Lock()
_pool = None # (workers, ThreadPoolExecutor)
_retired = [] # pools replaced by a new size, finishing what they hold
def refused_at_shutdown(error):
@@ -45,39 +63,113 @@ def note_refusal(logger, what):
logger.debug(f"{what}: refused, the interpreter is exiting")
def map_parallel(func, items, workers):
"""Apply ``func`` to every item on up to ``workers`` threads, keeping order.
def job_pool_started():
"""Whether a job pool exists (the teardown stops it)."""
return _pool is not None or bool(_retired)
``workers`` of 1 (or fewer) runs on the calling thread with no pool. An
exception in ``func`` propagates to the caller, so callers that must
survive bad input wrap ``func`` themselves. A pool refused at interpreter
exit runs the items it did not start on the calling thread (#375). An item
def _job_pool(workers):
"""The shared pool of ``workers`` threads, replacing one of another size."""
global _pool
with _pool_lock:
if _pool is not None and _pool[0] == workers:
return _pool[1]
retired = _pool
_pool = (
workers,
ThreadPoolExecutor(workers, thread_name_prefix=JOB_THREAD_PREFIX),
)
current = _pool[1]
if retired is not None:
retired[1].shutdown(wait=False) # it still runs what it holds
with _pool_lock:
_retired[:] = [p for p in _retired if _alive(p)] + [retired[1]]
return current
def shutdown_job_pool(timeout=None):
"""Stop the job pool (and any pool a resize retired): queued items are
cancelled (their callers run them on their own thread), running ones are
joined up to ``timeout`` seconds. True when no job thread is left."""
global _pool
with _pool_lock:
current, _pool = _pool, None
executors = _retired[:] + ([] if current is None else [current[1]])
_retired.clear()
threads = []
for executor in executors:
# After shutdown: it takes the lock ``submit`` holds while it starts
# and records a thread, so every thread is listed and none starts.
executor.shutdown(wait=False, cancel_futures=True)
threads.extend(getattr(executor, "_threads", ()))
deadline = None if timeout is None else time.monotonic() + timeout
for thread in threads:
thread.join(None if deadline is None else max(0.0, deadline - time.monotonic()))
return not any(thread.is_alive() for thread in threads)
def _alive(executor):
return any(t.is_alive() for t in getattr(executor, "_threads", ()))
def _forget(executor):
"""Drop *executor* as the shared pool (a later call starts a new one)."""
global _pool
with _pool_lock:
if _pool is not None and _pool[1] is executor:
_pool = None
def _on_job_thread():
return threading.current_thread().name.startswith(JOB_THREAD_PREFIX + "_")
def map_parallel(func, items, workers):
"""Apply ``func`` to every item on the shared pool of ``workers``
threads, keeping order.
``workers`` below 1 counts as 1. A call made from a job thread runs on
that thread, so nested work never waits for its own pool. An exception
in ``func`` propagates to the caller once the call's other items have
stopped (those not started are dropped), so callers that must survive
bad input wrap ``func`` themselves. A pool refused at interpreter exit
runs the items it did not start on the calling thread (#375). An item
that has not started when the interpreter starts exiting raises
ExitRequested instead (#376).
"""
func = checked(func)
items = list(items)
workers = max(1, int(workers or 1))
if workers == 1 or len(items) <= 1:
if not items or _on_job_thread():
return [func(item) for item in items]
max_workers = min(workers, len(items))
with ThreadPoolExecutor(max_workers, thread_name_prefix="pm-worker") as pool:
futures = []
try:
for item in items:
futures.append(pool.submit(func, item))
except RuntimeError as error:
if not refused_at_shutdown(error):
raise
pool = _job_pool(max(1, int(workers or 1)))
futures = []
try:
for item in items:
futures.append(pool.submit(func, item))
except RuntimeError as error:
if refused_at_shutdown(error):
note_refusal(_logger, "A worker pool")
# Drop what was queued but not started; it runs here instead.
# (A refused Thread.start leaves its item queued without a future
# here: it is cancelled too, unless a busy pool thread took it in
# between, the one case where an item can run twice.)
pool.shutdown(wait=True, cancel_futures=True)
else:
return [future.result() for future in futures]
return _run_rest(func, items, futures)
_forget(pool)
pool.shutdown(wait=False, cancel_futures=True)
elif SHUT_POOL_REFUSAL not in str(error):
raise
return _collect(func, items, futures)
def _collect(func, items, futures):
"""``_run_rest``; on an error, first drop the call's items not started
and wait for its running ones, as the old per-call pool's exit did."""
try:
return _run_rest(func, items, futures)
except BaseException:
for future in futures:
future.cancel()
wait(futures)
raise
def _run_rest(func, items, futures):
+116
View File
@@ -0,0 +1,116 @@
"""PNG text chunks without decoding pixels (#523).
Pillow's ``PngImageFile.text`` calls ``load()``, which decodes every pixel, to
reach text chunks written after the image data. On a scan or a busy image
monitor that is a full-size buffer per file for a few kilobytes of text.
``png_text`` returns the same dictionary as ``image.text`` for an opened,
single-frame PNG, by walking the chunks after the first IDAT with Pillow's own
chunk handlers (same decoding, limits and duplicate rule) and seeking over the
pixel data. It follows Pillow's ``load_end`` loop: chunk checksums after the
pixels are not checked, an unreadable chunk header ends the walk, and another
chunk that runs past the end of the file fails like ``ImageFile._safe_read``
(unless ``ImageFile.LOAD_TRUNCATED_IMAGES``).
The one difference is pixel data, which is never decoded: pixel data that is
undecodable, or cut short, no longer hides the text read before it. Pillow
itself accepts a file cut inside its last pixel chunk once the decoder has
all rows, and only a decode can tell those apart, so the walk never reads
less than ``.text`` did (the PR #170 "legacy parity floor").
Anything else (other formats, animated PNGs, an image already loaded) is
answered by Pillow itself.
"""
import os
import struct
from PIL import ImageFile
_PIXEL_CHUNKS = (b"IDAT", b"fdAT")
# Signature, IHDR chunk and IDAT chunk header: the earliest pixel data offset.
_FIRST_DATA_OFFSET = 8 + 25 + 8
def png_text(image) -> dict:
"""``dict(image.text)`` for an opened image, without a pixel decode.
The image stays usable: its file position and Pillow state are restored.
Non-PNG images answer ``{}`` (only PNG has text chunks in Pillow).
"""
if not _walkable(image):
return dict(getattr(image, "text", None) or {})
fp = image.fp
resume = fp.tell()
try:
return _walk_after_pixels(image.png, fp, image.tile[0][2] - 8)
finally:
fp.seek(resume)
def _walkable(image) -> bool:
"""A plain single-frame PNG whose header pass stopped at its first IDAT."""
tile = getattr(image, "tile", None)
return (
image.format == "PNG"
and getattr(image, "png", None) is not None
and getattr(image, "fp", None) is not None
and not getattr(image, "is_animated", False)
and isinstance(tile, list)
and len(tile) == 1
and tile[0][0] == "zip"
# The first IDAT's data offset; anything else is not a layout we know.
and isinstance(tile[0][2], int)
and tile[0][2] >= _FIRST_DATA_OFFSET
)
def _walk_after_pixels(stream, fp, offset: int) -> dict:
"""Pillow's ``load_end`` chunk loop from the first IDAT header, skipping pixels.
``stream`` is the image's own ``PngStream``: its text so far and its text
memory budget carry on exactly as a ``load()`` would carry them. The text
and info dictionaries are copied first, so ``image.text`` and
``image.info`` are unchanged. Like ``load_end``, a handler for a chunk
misplaced after the pixels (a second IHDR or PLTE) may still update the
stream's other header fields; Pillow ignores them once the image is open.
"""
end = fp.seek(0, os.SEEK_END)
saved = stream.im_text, stream.im_info, stream.text_memory
stream.im_text, stream.im_info = dict(saved[0]), dict(saved[1])
try:
fp.seek(offset)
while True:
try:
kind, position, length = stream.read()
except (struct.error, SyntaxError):
break
if kind == b"IEND":
break
try:
handled = kind not in _PIXEL_CHUNKS and _handle(
stream, kind, position, length
)
except UnicodeDecodeError:
break
if not handled:
if position + length > end:
if kind in _PIXEL_CHUNKS or ImageFile.LOAD_TRUNCATED_IMAGES:
break
raise OSError("Truncated File Read")
fp.seek(position + length + 4)
continue
fp.read(4) # checksum: not verified after the pixels, as in Pillow
return stream.im_text
finally:
stream.im_text, stream.im_info, stream.text_memory = saved
def _handle(stream, kind: bytes, position: int, length: int) -> bool:
"""Run Pillow's handler; ``False`` when the chunk's bytes must be skipped."""
try:
stream.call(kind, position, length)
except (EOFError, AttributeError):
# A chunk Pillow reads past (EOFError) or has no handler for.
return False
return True
+36 -9
View File
@@ -11,6 +11,9 @@
* - Results (`section.surface`): the outcome list (`button[aria-pressed]` with counts, in the mockup's order), a path
* filter, and the selected outcome's listed rows (at most 200 per outcome, so "Showing N of M"). Rows under Added and
* Already known open the file in the existing viewer through its listing identity (D57); the others are plain text.
* - #550: a row with no path is a record (a maintenance apply's "Prompt 12: category …"): its detail is shown as plain
* text, never opened; the filter reads it. A maintenance plan of records (every operation but Missing-file links)
* says "records" where a file job says "files" (`groupCopy`).
* - Log: `details.term` "Scan log", collapsed, holding the raw timestamped lines in `pre[tabindex=0]`.
* - D89: the tiles, the outcome list, the openable rows and the log and cancel labels are the job kind's (`jobView`):
* an image tagging job shows Tagged, Unchanged, Skipped and Errors, opens Tagged and Unchanged rows, and reads
@@ -44,6 +47,10 @@ export const DETAIL_INTERVAL_MS = 1000;
export const NOT_FOUND = 'This job is not in this session’s Activity. Jobs are kept until ComfyUI restarts.';
export const EMPTY_GROUP = 'No files in this group.';
export const NO_MATCH = 'No listed file matches this filter.';
export const EMPTY_RECORDS = 'No records in this group.';
export const NO_RECORD_MATCH = 'No listed record matches this filter.';
const FILE_FILTER = 'Filter by file path…';
const RECORD_FILTER = 'Filter records…';
export const VIEWER_HINT = 'each row opens the image in the viewer';
export const FILE_GONE = 'This file is no longer in Images. Rescan from Import and try again.';
export const OPEN_FAILED = 'The file could not be opened. Try again.';
@@ -68,10 +75,11 @@ export function parseDetail(raw) {
const groups = raw.outcomes !== null && typeof raw.outcomes === 'object' ? raw.outcomes : {};
for (const [group, rows] of Object.entries(groups)) {
if (!Array.isArray(rows)) continue;
outcomes[group] = rows.filter(row => row && isText(row.path) && row.path)
// A file row has a path; a record row (#550) has none, only its detail.
outcomes[group] = rows.filter(row => row && isText(row.path) && (row.path || (isText(row.detail) && row.detail)))
.map(row => ({ path: row.path, root: isText(row.root) ? row.root : '',
root_index: Number.isSafeInteger(row.root_index) && row.root_index >= 0 ? row.root_index : null,
detail: isText(row.detail) ? row.detail : '' }));
detail: isText(row.detail) ? row.detail : '', record: !row.path }));
}
const log = Array.isArray(raw.log) ? raw.log.filter(isText) : [];
return { job, outcomes, log };
@@ -129,11 +137,23 @@ export function resultsNote(shown, total, openable) {
return parts.length === 1 && openable && shown >= total ? `${text[0].toUpperCase()}${text.slice(1)}.` : text;
}
/** Rows whose path (or root label and path) holds the case-folded filter. */
/** Rows whose path (or root label and path), or a record's detail (#550), holds the case-folded filter. */
export function filterRows(rows, filter) {
const wanted = filter.trim().toLocaleLowerCase();
if (!wanted) return rows;
return rows.filter(row => `${row.root}/${row.path}`.toLocaleLowerCase().includes(wanted));
return rows.filter(row => (row.record ? row.detail : `${row.root}/${row.path}`).toLocaleLowerCase().includes(wanted));
}
/** #550: whether a job lists records (a maintenance plan of records), not files. */
export function listsRecords(job) {
return job?.kind === 'maintenance' && job.scope?.operation !== 'missing_links';
}
/** The empty-group, no-match and filter copy: "records" for a records job, "files" otherwise (#550). */
export function groupCopy(job) {
return listsRecords(job)
? { empty: EMPTY_RECORDS, noMatch: NO_RECORD_MATCH, placeholder: RECORD_FILTER }
: { empty: EMPTY_GROUP, noMatch: NO_MATCH, placeholder: FILE_FILTER };
}
/** The first outcome with files, so a view opens on what happened; the first outcome when nothing has yet. */
@@ -176,7 +196,7 @@ export function mountActivityJobView(main, { store, scope, api, jobId }) {
const search = element('label', undefined, 'search-field');
const input = element('input');
input.type = 'search';
input.placeholder = 'Filter by file path…';
input.placeholder = FILE_FILTER;
input.setAttribute('aria-label', 'Filter results');
search.append(input);
resultsPart.heading.append(search);
@@ -280,9 +300,13 @@ export function mountActivityJobView(main, { store, scope, api, jobId }) {
// Listed rows only ever grow, so a row keeps its node across reads: a focused row (or the viewer's trigger) stays.
let rowNodes = new Map();
function rowNode(row, index, openable) {
const key = `${selected}\n${index}\n${row.root}\n${row.path}`;
const key = `${selected}\n${index}\n${row.root}\n${row.path}\n${row.record ? row.detail : ''}`;
let node = rowNodes.get(key);
if (!node) {
if (!node && row.record) {
// #550: a record is its detail, plain text, with no open or locate action.
node = element('div', undefined, 'file-row record-row');
node.append(element('span', row.detail, 'why'));
} else if (!node) {
node = openable ? createButton('', 'file-row') : element('div', undefined, 'file-row');
node.append(element('span', row.path, 'path'), element('span', row.detail, 'why'));
if (openable) node.row = row;
@@ -300,8 +324,11 @@ export function mountActivityJobView(main, { store, scope, api, jobId }) {
const nodes = [...kept.values()];
// The counter is exact; at most 200 of its files are listed, and the filter narrows those.
const total = Math.max(job.counters[selected] ?? 0, rows.length);
if (!nodes.length) nodes.push(element('p', rows.length ? NO_MATCH : EMPTY_GROUP, 'empty-state'));
const note = wanted.length ? resultsNote(wanted.length, total, openable) : '';
const copy = groupCopy(job);
input.placeholder = copy.placeholder;
if (!nodes.length) nodes.push(element('p', rows.length ? copy.noMatch : copy.empty, 'empty-state'));
const opens = openable && wanted.some(row => !row.record);
const note = wanted.length ? resultsNote(wanted.length, total, opens) : '';
if (note) nodes.push(element('p', note, 'small'));
for (const child of [...files.children]) if (!nodes.includes(child)) child.remove();
nodes.forEach((node, index) => {
+100 -15
View File
@@ -31,6 +31,26 @@ const SOURCES = new Set(['images', 'library', 'review', 'tags', 'bulk']);
const OPTION_KEYS = new Set(['general_threshold', 'character_threshold', 'instructions', 'suggestion_limit']);
const DEFAULTS = Object.freeze({general_threshold: 0.35, character_threshold: 0.85, instructions: '', suggestion_limit: 10});
const MAX_SUGGESTIONS = 100;
/**
* #431: Built-in's text (`DEFAULT_TAG_TEMPLATE` in py/tagging_defaults.py, pinned equal by
* tests/test_saved_instructions.py), for when `/autotag/models` cannot be read.
*/
export const BUILTIN_INSTRUCTIONS = 'Generate only comma-separated Danbooru tags (lowercase_underscores, no spaces).\n'
+ 'Include: subject_count (1girl/1boy/etc), gender, ethnicity (asian, caucasian, etc), age_approx (teen, young_woman), '
+ 'body_attributes, face_features, hairstyle, appearance, clothing, accessories, pose, expression, action, camera_angle, '
+ 'camera_shot, scene_composition (wide_shot, long_shot, dutch_angle), environment, background_elements, lighting, mood.\n'
+ 'Prioritize visual accuracy.\n'
+ 'No sentences, no commentary, no captions. Only tags. Keep length short but descriptive.';
export const BUILTIN_ID = 'builtin';
/** The saved instructions when the server cannot say (D204): Built-in only, the default of both engines. */
const FALLBACK_CHOICES = Object.freeze({
entries: Object.freeze([Object.freeze({id: BUILTIN_ID, name: 'Built-in', text: BUILTIN_INSTRUCTIONS})]),
defaults: Object.freeze({gguf: BUILTIN_ID, hf: BUILTIN_ID}),
});
/** The value and label of the select's extra option while the text differs from the chosen entry (D204). */
export const EDITED_VALUE = 'edited';
export const EDITED_LABEL = 'Edited for this run';
export const EDITED_NOTE = 'Edits apply to this run only. Manage saved instructions in System › Auto-tagging.';
const MAX_INSTRUCTION_CHARS = 4000;
const EDITABLE = new Set(['ready', 'generate-error', 'write-error', 'conflict']);
// Answers after which only a fresh observation can save (the capture or epoch is gone); Retry observes again (D116).
@@ -182,27 +202,54 @@ function validateReceipt(value) {
return value;
}
/** #431: the saved instructions of `/autotag/models` (Built-in first) and each engine's default, else Built-in only. */
export function readChoices(value) {
const entries = Array.isArray(value?.instructions) ? value.instructions.filter(entry => object(entry)
&& typeof entry.id === 'string' && typeof entry.name === 'string' && typeof entry.text === 'string') : [];
if (entries[0]?.id !== BUILTIN_ID) return FALLBACK_CHOICES;
const ids = new Set(entries.map(entry => entry.id));
const defaults = Object.fromEntries(['gguf', 'hf'].map(engine => {
const id = value?.default_instructions?.[engine];
return [engine, ids.has(id) ? id : BUILTIN_ID];
}));
return Object.freeze({entries: Object.freeze(entries.map(entry => Object.freeze({...entry}))), defaults: Object.freeze(defaults)});
}
/** The language-model engine whose default a model reads (GGUF for a WD14 model, which takes none). */
const engineOf = model => (model === 'hf' ? 'hf' : 'gguf');
/** The default entry of `model`'s engine. */
export const engineDefault = (choices, model) => choices.entries.find(entry => entry.id === choices.defaults[engineOf(model)])
?? choices.entries[0];
/** The entry to show for `text` (a preset): the one with that text, else the engine's default. */
const entryFor = (choices, model, text) => choices.entries.find(entry => entry.text === text) ?? engineDefault(choices, model);
/** What the instruction fields hold for `model` with its engine's default (#431). */
const instructionsFor = (choices, model) => {
const entry = engineDefault(choices, model);
return {instructionId: entry.id, text: entry.text};
};
/**
* Thresholds, instructions and suggestion count from the server, else the documented defaults, the models'
* download status (`models`, null when unknown) and the saved tagging model (`model`, null when unknown; #328).
* download status (`models`, null when unknown), the saved tagging model (`model`, null when unknown; #328) and the
* saved instructions (`choices`, #431).
*/
async function readDefaults(api, signal) {
try {
const value = await api.get(MODELS_URL, {signal, cache: 'no-store'});
if (!object(value)) return {defaults: DEFAULTS, models: null, model: null};
if (!object(value)) return {defaults: DEFAULTS, models: null, model: null, choices: FALLBACK_CHOICES};
return {
choices: readChoices(value),
model: typeof value.model_type === 'string' && modelOf(value.model_type) ? value.model_type : null,
defaults: {
general_threshold: unit(value.wd14_general_threshold) ? value.wd14_general_threshold : DEFAULTS.general_threshold,
character_threshold: unit(value.wd14_character_threshold) ? value.wd14_character_threshold : DEFAULTS.character_threshold,
instructions: typeof value.default_prompt === 'string' ? value.default_prompt : DEFAULTS.instructions,
suggestion_limit: limit(value.suggestion_limit) ? value.suggestion_limit : DEFAULTS.suggestion_limit,
},
models: object(value.models) ? value.models : null,
};
} catch {
// The defaults keep the dialog usable; Generate reports a real failure itself.
return {defaults: DEFAULTS, models: null, model: null};
return {defaults: DEFAULTS, models: null, model: null, choices: FALLBACK_CHOICES};
}
}
@@ -258,6 +305,8 @@ function initialState(source, preset) {
model: preset?.model ?? MODEL_OPTIONS[0].id, options: {...DEFAULTS, ...preset?.options}, generated: false, suggestions: [],
suggestionsText: COPY.modelChanged, statusText: '', generateSupported: false, generateReason: null, toast: null,
skippedCount: 0,
// #431 (D204): the saved instructions, each engine's default and the entry the select shows.
instructionChoices: FALLBACK_CHOICES, instructionId: BUILTIN_ID,
// Bumped whenever the controller itself rewrites the option fields or the manual draft, never while typing.
optionsRevision: 0, draftRevision: 0,
};
@@ -269,6 +318,13 @@ export function skippedNote(count) {
return count === 1 ? ' 1 skipped tag was not suggested.' : ` ${count.toLocaleString('en-US')} skipped tags were not suggested.`;
}
/** D204: whether a language model's instructions differ from the chosen saved entry's text (a one-off edit). */
export function instructionsEdited(state) {
if (modelOf(state.model)?.classifier !== false) return false;
const entry = state.instructionChoices?.entries.find(item => item.id === state.instructionId);
return !entry || entry.text !== state.options.instructions;
}
/** Derived labels and permissions; never stored independently of the fields they follow. */
function derive(state) {
const editable = EDITABLE.has(state.phase);
@@ -277,6 +333,7 @@ function derive(state) {
primaryLabel: state.generated ? 'Approve selected tags' : 'Save manual tags',
countText: state.generated ? `${state.suggestions.length} ${noun(state.suggestions.length, 'suggestion')}` : 'Not generated',
chipHint: state.suggestions.length ? COPY.chipHint + skippedNote(state.skippedCount) : '',
instructionsEdited: instructionsEdited(state),
canSave: editable,
canGenerate: editable && state.generateSupported,
canRetry: RETRYABLE.has(state.phase),
@@ -394,6 +451,10 @@ export function createImageTaggingController({api, target, source = 'images', on
if (closed || serial !== openSerial) return;
// The saved model (System › Auto-tagging) unless a preset or the user already chose one (#328).
const savedModel = !chosen && !modelPicked && read.model ? {model: read.model} : {};
// #431: the instructions start from the model's engine default; a preset's text wins (D204).
const model = savedModel.model ?? state.model;
const presetText = chosen?.options.instructions;
const entry = presetText === undefined ? engineDefault(read.choices, model) : entryFor(read.choices, model, presetText);
observedOnce = true;
const {width, height, media_type: mediaType, preview_url: previewUrl, generate} = value.capture;
// The draft is read now, so text typed while observing is kept (#515). A kept draft stays as typed unless
@@ -410,7 +471,8 @@ export function createImageTaggingController({api, target, source = 'images', on
savedTags: value.tags.slice(), tagRevision: value.tag_revision,
manual: draft === null ? value.tags.join(', ') : changed || unseen ? mergeCurrent(value.tags, draft) : draft,
draftRevision: state.draftRevision + 1,
...savedModel, options: {...defaults, ...chosen?.options}, optionsRevision: state.optionsRevision + 1,
...savedModel, options: {...defaults, instructions: entry.text, ...chosen?.options}, optionsRevision: state.optionsRevision + 1,
instructionChoices: read.choices, instructionId: entry.id,
generateSupported: generate.supported, generateReason: generate.reason,
statusText: changed ? `${COPY.changedElsewhere} ${savedNow(value.tags)}${reason ? ` ${reason}` : ''}` : reason,
});
@@ -428,15 +490,26 @@ export function createImageTaggingController({api, target, source = 'images', on
if (id === state.model) return;
stopGenerating();
const phase = state.phase === 'generating' || state.phase === 'generate-error' ? 'ready' : state.phase;
const {instructionId, text} = instructionsFor(state.instructionChoices, id);
set({
phase, model: id, generated: false, suggestions: [], suggestionsText: COPY.modelChanged, statusText: reasonText(),
optionsRevision: state.optionsRevision + 1,
// Like the mock, a model change re-renders its fields with the defaults; the count is kept (D91).
// Like the mock, a model change re-renders its fields with the defaults; the count is kept (D91). The
// instructions start from the new engine's default, dropping a one-off edit (D204).
options: {...state.options, general_threshold: defaults.general_threshold,
character_threshold: defaults.character_threshold, instructions: defaults.instructions},
character_threshold: defaults.character_threshold, instructions: text},
instructionId,
});
}
/** D204: choose a saved entry; its text replaces the instructions (and a one-off edit). */
function pickInstructions(id) {
if (closed) return;
const entry = state.instructionChoices.entries.find(item => item.id === id);
if (!entry) return;
set({instructionId: id, options: {...state.options, instructions: entry.text}, optionsRevision: state.optionsRevision + 1});
}
function setOption(key, value) {
requireValue(OPTION_KEYS.has(key), 'Unknown tagging option');
set({options: {...state.options, [key]: value}});
@@ -593,7 +666,7 @@ export function createImageTaggingController({api, target, source = 'images', on
return {
get state() {return state;},
get target() {return imageTarget;},
subscribe, open, setManual, setModel, setOption, generate, remove, save, retry, retarget, cancel, dispose,
subscribe, open, setManual, setModel, setOption, pickInstructions, generate, remove, save, retry, retarget, cancel, dispose,
};
}
@@ -612,8 +685,9 @@ export function createTaggingOptionsController({api} = {}) {
let abort = null;
let modelPicked = false;
const deriveOptions = value => ({...value, canGenerate: false, unavailable: modelUnavailable(models, value.model),
canStart: value.phase === 'ready' && !modelUnavailable(models, value.model)});
let state = freeze(deriveOptions({phase: 'loading', model: MODEL_OPTIONS[0].id, options: {...DEFAULTS}, optionsRevision: 0}));
canStart: value.phase === 'ready' && !modelUnavailable(models, value.model), instructionsEdited: instructionsEdited(value)});
let state = freeze(deriveOptions({phase: 'loading', model: MODEL_OPTIONS[0].id, options: {...DEFAULTS}, optionsRevision: 0,
instructionChoices: FALLBACK_CHOICES, instructionId: BUILTIN_ID}));
function set(patch) {
if (closed) return;
@@ -628,17 +702,28 @@ export function createTaggingOptionsController({api} = {}) {
({defaults, models} = read);
// The saved model (System › Auto-tagging) unless the user already chose one (#328).
const savedModel = !modelPicked && read.model ? {model: read.model} : {};
set({phase: 'ready', ...savedModel, options: {...defaults}, optionsRevision: state.optionsRevision + 1});
const {instructionId, text} = instructionsFor(read.choices, savedModel.model ?? state.model);
set({phase: 'ready', ...savedModel, options: {...defaults, instructions: text}, optionsRevision: state.optionsRevision + 1,
instructionChoices: read.choices, instructionId});
}
function setModel(id) {
requireValue(modelOf(id), 'Unknown tagging model');
if (closed) return;
modelPicked = true;
if (id === state.model) return;
// As in the Tag image dialog: the model's fields start from the defaults again; the count is kept (D91).
set({model: id, optionsRevision: state.optionsRevision + 1,
// As in the Tag image dialog: the model's fields start from the defaults again; the count is kept (D91), and
// the instructions from the new engine's default (D204).
const {instructionId, text} = instructionsFor(state.instructionChoices, id);
set({model: id, optionsRevision: state.optionsRevision + 1, instructionId,
options: {...state.options, general_threshold: defaults.general_threshold,
character_threshold: defaults.character_threshold, instructions: defaults.instructions}});
character_threshold: defaults.character_threshold, instructions: text}});
}
/** D204: choose a saved entry; its text replaces the instructions. */
function pickInstructions(id) {
if (closed) return;
const entry = state.instructionChoices.entries.find(item => item.id === id);
if (!entry) return;
set({instructionId: id, options: {...state.options, instructions: entry.text}, optionsRevision: state.optionsRevision + 1});
}
function setOption(key, value) {
requireValue(OPTION_KEYS.has(key), 'Unknown tagging option');
@@ -661,6 +746,6 @@ export function createTaggingOptionsController({api} = {}) {
}
return {
get state() {return state;},
subscribe, open, setModel, setOption, request, dispose,
subscribe, open, setModel, setOption, pickInstructions, request, dispose,
};
}
+47 -3
View File
@@ -17,7 +17,8 @@
* flight at close keeps its capture until it settles, and its outcome is still toasted and announced.
*/
import { createButton, createScope } from '../shared/toolkit/core.js';
import { createImageTaggingController, MODEL_OPTIONS, optionError } from './image-tagging-state.mjs';
import { createImageTaggingController, EDITED_LABEL, EDITED_NOTE, EDITED_VALUE, MODEL_OPTIONS, engineDefault, optionError }
from './image-tagging-state.mjs';
import { closeWorkspaceDialog, field, openContentDialog } from './workspace-dialogs.mjs';
import { createWorkspaceIcon } from './icons.mjs';
import { showWorkspaceToast } from './toast.mjs';
@@ -107,9 +108,13 @@ function numberInput(value, {min, max, step}) {
* It follows `controller` until `scope` is disposed (reused by the bulk Auto-tag dialog, D85, without Generate:
* `generate: false`). Returns the `element`, `validate()` (the fields' native validity, then what the controller would
* refuse; true when valid, else the first invalid field reports) and `requestGenerate()`, Generate's own path (field
* validity first), for a Retry after a failed Generate.
* validity first), for a Retry after a failed Generate. D204 (#431): for a language model, a "Saved instructions" select
* (`pickId`) sits above "Tagging instructions": it starts at the engine's default ("<name> · default"), a pick replaces
* the text, and while the text differs from the picked entry it shows "Edited for this run" (not choosable) with a
* muted note under the textarea.
*/
export function renderModelPanel({scope, controller, heading = 'Auto-tag this image', generate: withGenerate = true}) {
export function renderModelPanel({scope, controller, heading = 'Auto-tag this image', generate: withGenerate = true,
pickId = 'image-tag-instructions-pick'}) {
const panel = el('div', undefined, 'tag-model-panel');
const select = el('select');
for (const option of MODEL_OPTIONS) {
@@ -125,9 +130,35 @@ export function renderModelPanel({scope, controller, heading = 'Auto-tag this im
if (withGenerate) panel.append(generate);
const inputs = new Map();
let shownModel = null, shownOptions = null;
// D204: the Saved instructions select and the one-off note, while a language model is chosen.
let pick = null, editedNote = null, shownEdited = null;
/** The select's options: every entry (the engine default marked), plus "Edited for this run" while edited. */
function fillPick(state) {
shownEdited = state.instructionsEdited;
const fallback = engineDefault(state.instructionChoices, state.model).id;
pick.replaceChildren(...state.instructionChoices.entries.map(entry => {
const option = el('option', entry.id === fallback ? `${entry.name} · default` : entry.name);
option.value = entry.id;
return option;
}));
if (state.instructionsEdited) {
const option = el('option', EDITED_LABEL);
option.value = EDITED_VALUE;
option.disabled = true;
pick.append(option);
}
pick.value = state.instructionsEdited ? EDITED_VALUE : state.instructionId;
editedNote.hidden = !state.instructionsEdited;
// The note describes the textarea while it shows.
const area = inputs.get('instructions');
if (state.instructionsEdited) area?.setAttribute('aria-describedby', editedNote.id);
else area?.removeAttribute('aria-describedby');
}
function renderOptions(state) {
inputs.clear();
pick = editedNote = null;
const classifier = MODEL_OPTIONS.find(option => option.id === state.model).classifier;
const rows = [];
if (classifier) {
@@ -139,10 +170,17 @@ export function renderModelPanel({scope, controller, heading = 'Auto-tag this im
}
rows.push(pair);
} else {
pick = el('select');
pick.addEventListener('change', () => { if (pick.value !== EDITED_VALUE) controller.pickInstructions(pick.value); });
rows.push(field('Saved instructions', pick, pickId));
const area = el('textarea');
area.value = state.options.instructions;
inputs.set('instructions', area);
rows.push(field('Tagging instructions', area, OPTION_IDS.instructions));
editedNote = el('p', EDITED_NOTE, 'small instructions-edited');
editedNote.id = `${pickId}-note`;
rows.push(editedNote);
fillPick(state);
}
// D91: the number of suggestions, for every model.
const count = numberInput(state.options.suggestion_limit, {min: '1', max: '100', step: '1'});
@@ -162,7 +200,13 @@ export function renderModelPanel({scope, controller, heading = 'Auto-tag this im
shownModel = state.model;
shownOptions = state.optionsRevision;
select.value = state.model;
// A pick re-renders the fields; keyboard focus stays on the select it was made with.
const picking = pick !== null && document.activeElement === pick;
renderOptions(state);
if (picking) pick?.focus();
} else if (pick && state.instructionsEdited !== shownEdited) {
// Typing never re-renders the fields (I1): only the select and the note follow an edit.
fillPick(state);
}
setDisabled(generate, !state.canGenerate);
}
+2 -1
View File
@@ -167,7 +167,8 @@ export function openAutoTag({api, scope, store, images, trigger, onTagged = () =
const options = createTaggingOptionsController({api});
// D85: the Tag image dialog's model panel without Generate; its Suggestions field is the job's count (D91).
const {element: panel, validate} = renderModelPanel({scope: local, controller: options,
heading: `Auto-tag ${countOf(filter ? count : listed.length, 'image')}`, generate: false}); // D118
heading: `Auto-tag ${countOf(filter ? count : listed.length, 'image')}`, generate: false, // D118
pickId: 'bulk-tag-instructions-pick'}); // D204
let choice = APPLY_CHOICES[0];
const actions = el('div', undefined, 'dialog-actions');
const cancel = createButton('Cancel');
+2
View File
@@ -49,6 +49,8 @@ export const OPERATION_TITLES = Object.freeze({
duplicate_prompts: 'Duplicate prompt text',
optimize: 'Optimize database',
unused_components: 'Unused prompt components',
// D192 (#550): beyond the concept, System › Maintenance › Database.
pair_metadata: 'Pair category and tags',
missing_links: 'Missing-file links',
duplicate_files: 'Duplicate image files',
// Reviewed and applied through the API and seen in Activity; Maintenance has no row for it (the concept has none).
+150 -38
View File
@@ -11,6 +11,13 @@ const linkedTo = (target,roles) => Object.keys(roles).filter(role=>Array.isArray
const nodeLabel = target => `#${target.id} ${target.label}`;
// A lost reply leaves the outcome unknown; anything else the host said is a refusal, and nothing was written.
const LOST_REPLY = /timed out|aborted|disposed/i;
const outcomeUnknown = (posted,error) => posted && (LOST_REPLY.test(error.message) || /unexpected acknowledgement/.test(error.message));
const UNKNOWN_ADVICE = 'Check ComfyUI, then preview again.';
/** The Destination node value that sends to every eligible node of the window (#542, D191). Node ids never take it. */
export const ALL_NODES = '*';
// "All" sends in node id order: 2 before 10, and a subgraph id "3:1" after "3".
const byNodeId = (a,b) => a.id.localeCompare(b.id,'en',{numeric:true});
const allMessage = (count,where) => `Sending to all ${count} eligible nodes in ${where}.`;
/** Why nothing in the discovered windows can receive these roles, most specific first (D173). */
function unavailableMessage(found,roles) {
@@ -40,23 +47,90 @@ function distinctLabels(items,suffix) {
export function createSendSession({client, roles:rawRoles, onChange}) {
const roles = Object.freeze(validateRoles(rawRoles));
let state = {phase:'idle',hosts:[],hostId:'',targetId:'',preview:null,outcomeUnknown:false,message:''};
// `previews` holds the reviewed nodes of "All", in send order; `preview` the one node otherwise.
let state = {phase:'idle',hosts:[],hostId:'',targetId:'',preview:null,previews:[],outcomeUnknown:false,message:''};
let operation, generation = 0, disposed = false;
const emit = patch => { if (!disposed) { state = Object.freeze({...state,...patch}); onChange(state); } };
function invalidate() { generation++; operation?.abort(); operation = null; }
function begin(phase) { invalidate(); operation = new AbortController(); emit({phase,preview:null,outcomeUnknown:false,message:''}); return {signal:operation.signal,version:generation}; }
function begin(phase) { invalidate(); operation = new AbortController(); emit({phase,preview:null,previews:[],outcomeUnknown:false,message:''}); return {signal:operation.signal,version:generation}; }
const current = request => !disposed && generation === request.version;
const selected = () => state.hosts.find(host=>host.hostId===state.hostId)?.targets.find(target=>target.id===state.targetId);
async function fetchPreview(request) {
const preview=await client.preview(state.hostId,state.targetId,roles,{signal:request.signal});
const chosenHost = () => state.hosts.find(host=>host.hostId===state.hostId);
/** The nodes the current choice sends to: one, every eligible node of the window for "All", or none. */
function chosen() {
const host=chosenHost();
if (!host) return [];
if (state.targetId===ALL_NODES) return host.targets.length>1 ? host.targets.toSorted(byNodeId) : [];
const target=host.targets.find(item=>item.id===state.targetId);
return target ? [target] : [];
}
// checkRevision is off only for a later node of "All", whose revision our own earlier write may have moved.
async function fetchPreview(request,hostId,target,checkRevision=true) {
const preview=await client.preview(hostId,target.id,roles,{signal:request.signal});
if (!current(request)) return null;
const fields=preview?.fields;
if(typeof preview?.ticket!=='string' || !preview.ticket || preview.target?.id!==state.targetId || !Array.isArray(fields) || fields.length!==Object.keys(roles).length || new Set(fields.map(field=>field?.role)).size!==fields.length || Array.from(fields).some(field=>!field || !Object.hasOwn(roles,field.role) || typeof field.before!=='string' || field.after!==roles[field.role])) throw Error('ComfyUI returned an unexpected preview. Find the destination and preview again.');
if(typeof preview?.ticket!=='string' || !preview.ticket || preview.target?.id!==target.id || !Array.isArray(fields) || fields.length!==Object.keys(roles).length || new Set(fields.map(field=>field?.role)).size!==fields.length || Array.from(fields).some(field=>!field || !Object.hasOwn(roles,field.role) || typeof field.before!=='string' || field.after!==roles[field.role])) throw Error('ComfyUI returned an unexpected preview. Find the destination and preview again.');
// Undefined on a tab still running older code, which has nothing to compare.
const found=selected()?.revision;
if(found!==undefined && preview.revision!==found) throw Error('The workflow changed since this window was found. Nothing was sent. Find windows again.');
const found=target.revision;
if(checkRevision && found!==undefined && preview.revision!==found) throw Error('The workflow changed since this window was found. Nothing was sent. Find windows again.');
return preview;
}
function checkAcknowledgement(acknowledgement,targetId) {
if(acknowledgement?.targetId!==targetId || !acknowledgement.values || Object.keys(acknowledgement.values).length!==Object.keys(roles).length || Object.keys(roles).some(role=>!Object.hasOwn(acknowledgement.values,role) || acknowledgement.values[role]!==roles[role])) throw Error('ComfyUI returned an unexpected acknowledgement.');
}
const reviewedText = change => JSON.stringify([change.fields.map(field=>[field.role,field.before]).sort(),change.composition ?? null]);
const sameChanges = (reviewed,fresh) => reviewedText(reviewed)===reviewedText(fresh);
/** Q542-2 C: every node of "All" previews before anything is written; the first refusal names its node. */
async function previewEach(request,hostId,targets) {
const reviewed=[];
for (const target of targets) {
try {
const preview=await fetchPreview(request,hostId,target);
if (!preview) return null;
reviewed.push({target,label:target.label,preview});
} catch(error) {
if (!current(request)) return null;
throw Error(`${target.label}: ${error.message}${/Nothing was sent/.test(error.message) ? '' : ' Nothing was sent.'}`);
}
}
return reviewed;
}
/**
* Sends the reviewed nodes one after another, best effort past refusals. Only the first node of a pass taken in this
* click keeps its ticket: a held review may have expired at the host (30 s), and our own writes may make later
* tickets stale, so every other node takes a fresh ticket that must show exactly the reviewed text.
*/
async function applyEach(request,hostId,reviewed,where,held) {
const sent=[],refused=[];let written=false;
for (const [index,{target,preview}] of reviewed.entries()) {
let posted=false;
try {
let ticket=preview.ticket;
if (index>0 || held) {
// Until something was posted the revision is still the discovered one; after that our own write may move it.
const fresh=await fetchPreview(request,hostId,target,!written);
if (!fresh) return;
if (!sameChanges(preview,fresh)) throw Error('Its prompt changed since the preview. Preview again.');
ticket=fresh.ticket;
}
posted=written=true;
const acknowledgement=await client.apply(hostId,ticket,{signal:request.signal});
if (!current(request)) return;
checkAcknowledgement(acknowledgement,target.id);
sent.push(target);
} catch(error) {
if (!current(request)) return;
const summary=[`Sent to ${sent.length} of ${reviewed.length} nodes in ${where}.`,...refused];
if (outcomeUnknown(posted,error)) {
const unsent=reviewed.slice(index+1).map(item=>item.label);
emit({phase:'error',outcomeUnknown:true,message:[...summary,`No confirmation received from ${target.label}; the update may have applied.`,...(unsent.length ? [`Not sent: ${unsent.join(', ')}.`] : []),UNKNOWN_ADVICE,error.message].join(' ')});
return;
}
refused.push(`${target.label}: ${error.message}`);
}
}
const message=[`Sent to ${sent.length} of ${reviewed.length} nodes in ${where}.`,...refused].join(' ');
emit(refused.length ? {phase:'error',message} : {phase:'success',message});
}
function hostChoice(host) {
if (host.targets.length>1) return {hostId:host.hostId,targetId:'',message:`Choose a destination node in ${host.label}.`};
const [target] = host.targets;
@@ -79,38 +153,61 @@ export function createSendSession({client, roles:rawRoles, onChange}) {
else { const choice=hostChoice(hosts[0]); emit({phase:'select',hosts,...choice,message:choice.message+note}); }
} catch (error) { if (current(request)) emit({phase:'error',message:error.message}); }
},
selectHost(hostId) { if (disposed) return; invalidate(); const host=state.hosts.find(host=>host.hostId===hostId); emit({phase:'select',preview:null,outcomeUnknown:false,...(host ? hostChoice(host) : {hostId:'',targetId:'',message:'Choose a ComfyUI window.'})}); },
selectTarget(targetId) { if (disposed) return; invalidate(); const host=state.hosts.find(host=>host.hostId===state.hostId); const chosen=host?.targets.some(target=>target.id===targetId); emit({phase:'select',outcomeUnknown:false,targetId:chosen?targetId:'',preview:null,message:chosen?'Apply to send it, or preview the changes first.':'Choose a destination node, then preview changes.'}); },
selectHost(hostId) { if (disposed) return; invalidate(); const host=state.hosts.find(host=>host.hostId===hostId); emit({phase:'select',preview:null,previews:[],outcomeUnknown:false,...(host ? hostChoice(host) : {hostId:'',targetId:'',message:'Choose a ComfyUI window.'})}); },
selectTarget(targetId) {
if (disposed) return; invalidate(); const host=chosenHost();
// D173: "All" is a choice only where there are two or more nodes, and it is never made for you.
const all=targetId===ALL_NODES && host?.targets.length>1;
const chosenOne=all || Boolean(host?.targets.some(target=>target.id===targetId));
const message=all ? `${allMessage(host.targets.length,host.label)} Apply to send it, or preview the changes first.` : chosenOne ? 'Apply to send it, or preview the changes first.' : 'Choose a destination node, then preview changes.';
emit({phase:'select',outcomeUnknown:false,targetId:chosenOne?targetId:'',preview:null,previews:[],message});
},
async preview() {
if (disposed || !selected() || ['previewing','applying'].includes(state.phase)) return;
const targets=chosen();
if (disposed || !targets.length || ['previewing','applying'].includes(state.phase)) return;
const all=state.targetId===ALL_NODES; const hostId=state.hostId; const windowLabel=chosenHost()?.label;
const request=begin('previewing');
try {
const preview=await fetchPreview(request);
if (all) {
const previews=await previewEach(request,hostId,targets);
if (previews) emit({phase:'ready',previews,message:`${allMessage(targets.length,windowLabel)} Review the complete changes, then apply.`});
return;
}
const preview=await fetchPreview(request,hostId,targets[0]);
if (!preview) return;
const windowLabel=state.hosts.find(host=>host.hostId===state.hostId)?.label;
emit({phase:'ready',preview,message:`Sending to ${nodeLabel({id:preview.target.id,label:preview.target.label ?? ''})} in ${windowLabel}. Review the complete changes, then apply.`});
} catch(error) { if(current(request)) emit({phase:'error',preview:null,message:error.message}); }
} catch(error) { if(current(request)) emit({phase:'error',preview:null,previews:[],message:error.message}); }
},
async apply() {
// Success stays closed: the second click of a double click must not send again.
if(disposed || !selected() || !['select','ready','error'].includes(state.phase) || state.outcomeUnknown) return;
const held=state.phase==='ready' ? state.preview : null;
const targetId=state.targetId; const hostId=state.hostId;
const targets=chosen();
if(disposed || !targets.length || !['select','ready','error'].includes(state.phase) || state.outcomeUnknown) return;
const all=state.targetId===ALL_NODES;
const held=state.phase==='ready' ? (all ? state.previews : state.preview) : null;
const hostId=state.hostId; const windowLabel=chosenHost()?.label;
const request=begin('applying');
emit({message:'Waiting for ComfyUI confirmation. Closing this dialog does not undo a sent update.'});
if (all) {
let reviewed;
try { reviewed=held?.length ? held : await previewEach(request,hostId,targets); }
catch(error) { if(current(request)) emit({phase:'error',message:error.message}); return; }
if (reviewed) await applyEach(request,hostId,reviewed,windowLabel,Boolean(held?.length));
return;
}
const [target]=targets;
let posted=false;
try {
const preview=held ?? await fetchPreview(request);
const preview=held ?? await fetchPreview(request,hostId,target);
if(!preview || !current(request)) return;
posted=true;
const acknowledgement=await client.apply(hostId,preview.ticket,{signal:request.signal});
if(!current(request)) return;
if(acknowledgement?.targetId!==targetId || !acknowledgement.values || Object.keys(acknowledgement.values).length!==Object.keys(roles).length || Object.keys(roles).some(role=>!Object.hasOwn(acknowledgement.values,role) || acknowledgement.values[role]!==roles[role])) throw Error('ComfyUI returned an unexpected acknowledgement.');
checkAcknowledgement(acknowledgement,target.id);
emit({phase:'success',message:'Sent to ComfyUI.'});
} catch(error) {
if(!current(request)) return;
const unknown=posted && (LOST_REPLY.test(error.message) || /unexpected acknowledgement/.test(error.message));
emit({phase:'error',preview:null,outcomeUnknown:unknown,message:unknown ? `No confirmation received. The update may have applied. Check ComfyUI, then preview again. ${error.message}` : error.message});
const unknown=outcomeUnknown(posted,error);
emit({phase:'error',preview:null,outcomeUnknown:unknown,message:unknown ? `No confirmation received. The update may have applied. ${UNKNOWN_ADVICE} ${error.message}` : error.message});
}
},
dispose() { if(disposed) return; disposed=true; invalidate(); client.dispose(); },
@@ -159,7 +256,7 @@ export function createPromptSender({scope,baseUrl=window.ApiClient.resolveUrl('/
inspect.disabled=true;apply.disabled=true;
const actions=element('footer',undefined,'prompt-sender-actions');actions.append(find,inspect,apply);
dialog.append(header,description,controls,status,preview,actions);document.body.append(dialog);
let session,closed=false;
let session,closed=false,headingIds=0;
function dispose() {
if(closed) return;closed=true;local.dispose();session?.dispose();
if(dialog.open) dialog.close();dialog.remove();
@@ -176,8 +273,8 @@ export function createPromptSender({scope,baseUrl=window.ApiClient.resolveUrl('/
function options(select,placeholder,entries,value) {
select.replaceChildren(...[["",placeholder],...entries].map(([value,label])=>{const option=element('option',label);option.value=value;return option;}));select.value=value;
}
function textBlock(parent,label,value) {
parent.append(element('h4',label));const text=element('pre');text.tabIndex=0;text.append(element('code',value === '' ? '(empty)' : value));parent.append(text);
function textBlock(parent,label,value,level=4) {
parent.append(element(`h${Math.min(level,6)}`,label));const text=element('pre');text.tabIndex=0;text.append(element('code',value === '' ? '(empty)' : value));parent.append(text);
}
function render(state) {
if(closed) return;
@@ -187,28 +284,43 @@ export function createPromptSender({scope,baseUrl=window.ApiClient.resolveUrl('/
dialog.setAttribute('aria-busy',String(busy));
options(host,'Choose a ComfyUI window',state.hosts.map(item=>[item.hostId,item.label]),state.hostId);
const selected=state.hosts.find(item=>item.hostId===state.hostId);
options(target,'Choose a destination node',(selected?.targets || []).map(item=>[item.id,item.label]),state.targetId);
const nodes=(selected?.targets || []).map(item=>[item.id,item.label]);
// Q542-4 A: the last option sends to every eligible node of this window.
if(nodes.length>1) nodes.push([ALL_NODES,`All ${nodes.length} eligible nodes`]);
options(target,'Choose a destination node',nodes,state.targetId);
host.disabled=busy;target.disabled=busy || !selected;
find.disabled=busy;inspect.disabled=busy || !state.targetId;apply.disabled=busy || !state.targetId || state.phase==='success' || state.outcomeUnknown;
preview.replaceChildren();
if(state.preview) {
preview.append(element('h3','Review changes'));
for(const field of state.preview.fields || []) {
const section=element('section',undefined,`prompt-sender-field prompt-role-${field.role}`);
section.append(element('h3',field.role==='positive'?'Positive prompt':'Negative prompt'));
textBlock(section,'Before',field.before);textBlock(section,'After',field.after);preview.append(section);
}
const composition=state.preview.composition;
if(composition) {
const section=element('section',undefined,'prompt-sender-composition');section.append(element('h3','Positive composition'));
textBlock(section,'Prepend (preserved)',composition.prepend || '');textBlock(section,'Append (preserved)',composition.append || '');
textBlock(section,'Combined before',composition.before ?? 'Resolved at execution from connected inputs.');
textBlock(section,'Combined after',composition.after ?? 'Resolved at execution from connected inputs.');
if(composition.prependConnected || composition.appendConnected) section.append(element('p','Connected prepend and append inputs are preserved and resolved when the workflow runs.'));
preview.append(section);
changes(preview,state.preview,3);
}
if(state.previews?.length) {
preview.append(element('h3','Review changes'));
// One section per node of "All", in the order they will be sent, named by its heading.
for(const item of state.previews) {
const node=element('section',undefined,'prompt-sender-node');const heading=element('h4',item.label);
heading.id=`prompt-sender-node-${++headingIds}`;node.setAttribute('aria-labelledby',heading.id);
node.append(heading);changes(node,item.preview,5);preview.append(node);
}
}
}
function changes(parent,change,level) {
for(const field of change.fields || []) {
const section=element('section',undefined,`prompt-sender-field prompt-role-${field.role}`);
section.append(element(`h${level}`,field.role==='positive'?'Positive prompt':'Negative prompt'));
textBlock(section,'Before',field.before,level+1);textBlock(section,'After',field.after,level+1);parent.append(section);
}
const composition=change.composition;
if(composition) {
const section=element('section',undefined,'prompt-sender-composition');section.append(element(`h${level}`,'Positive composition'));
textBlock(section,'Prepend (preserved)',composition.prepend || '',level+1);textBlock(section,'Append (preserved)',composition.append || '',level+1);
textBlock(section,'Combined before',composition.before ?? 'Resolved at execution from connected inputs.',level+1);
textBlock(section,'Combined after',composition.after ?? 'Resolved at execution from connected inputs.',level+1);
if(composition.prependConnected || composition.appendConnected) section.append(element('p','Connected prepend and append inputs are preserved and resolved when the workflow runs.'));
parent.append(section);
}
}
dialog.showModal();
try {
const url=new URL(baseUrl,window.location.href);
+16 -2
View File
@@ -179,13 +179,27 @@ export function reviewRefusal(error) {
return REVIEW_FAILED;
}
/**
* #583: the review uploads not answered yet. WebKit can read a FormData's file part after the fetch call returned (it
* did whenever a request interceptor, here Playwright's, was armed); a garbage collection in between dropped a form
* nothing referenced any more, and the server got an empty file. Each form is held here until its upload is answered.
* Exported for the unit tests only.
*/
export const uploadsInFlight = new Set();
/** Stage `file` for review; the server's answer, or an ApiError naming its refusal. */
export async function uploadRestore(file, { fetchUrl = window.ApiClient.fetchUrl, ApiError = window.ApiClient.ApiError,
signal } = {}) {
const form = new FormData();
form.append('database_file', file, file.name);
const response = await fetchUrl(PREVIEW_URL, { method: 'POST', body: form, cache: 'no-store', credentials: 'same-origin',
signal });
uploadsInFlight.add(form);
let response;
try {
response = await fetchUrl(PREVIEW_URL, { method: 'POST', body: form, cache: 'no-store', credentials: 'same-origin',
signal });
} finally {
uploadsInFlight.delete(form);
}
let data = null;
try {
data = await response.json();
+7 -2
View File
@@ -2,10 +2,12 @@
* System › Maintenance (#249 System S7 Task 19): the concept's `renderSystem` "maintenance" body (app.js 688–690).
*
* - The notice "Check first. Review changes. Then apply." with "Run health checks" (D123), then the surfaces
* "Database" and "Images & cache" with the concept's eight operations, in its order, labels and danger marks.
* "Database" and "Images & cache" with the concept's eight operations, in its order, labels and danger marks, plus
* D192 (#550, beyond the concept): "Pair category and tags" after "Unused prompt components".
* - Integrity & relationship checks → Check runs the same read-only health check and opens the Health report (D95);
* its "Review missing links" reviews a fresh Missing-file links plan, never a cleanup.
* - Duplicate prompt text, Optimize database, Unused prompt components and Missing-file links → Review plan open the
* - Duplicate prompt text, Optimize database, Unused prompt components, Pair category and tags (`pair_metadata`: fill
* old Pair prompts' empty category and tags from their saved pairs) and Missing-file links → Review plan open the
* D100 plan review (maintenance-review.mjs), which applies only what it showed, as an Activity job.
* - Duplicate image files → Find duplicates starts the duplicate search as an Activity job and opens it; its view
* offers Review groups (D101) once it found some.
@@ -39,6 +41,9 @@ export const MAINTENANCE_SURFACES = Object.freeze([
['Optimize database', 'Reclaim unused SQLite space. This does not delete prompts or media.', 'Review plan', 'optimize'],
['Unused prompt components', 'Preview unreferenced components. Pair references and history remain protected.', 'Review plan',
'unused_components', true],
// D192 (#550): beyond the concept. Fills only empty fields, deletes nothing (no danger mark).
['Pair category and tags', 'Fill an empty category or tags on prompts older Pair node runs saved, from their saved ' +
'pairs. Only empty fields of never-edited prompts.', 'Review plan', 'pair_metadata'],
]],
['Images & cache', [
['Missing-file links', 'Remove stale database links only. Files on disk are never deleted by this operation.', 'Review plan',
+7
View File
@@ -416,6 +416,9 @@ pre.prompt-text { font: inherit; font-size: 12px; line-height: 1.75; margin-bloc
.prompt-sender-preview { margin-block: 20px; min-inline-size: 0; /* D48 */ }
.prompt-sender-field { border: 1px solid var(--ca-line); border-radius: 7px; padding: 14px; margin-block: 12px; min-inline-size: 0; /* D48 */ }
.prompt-sender-field :where(pre, textarea), .prompt-sender-composition { white-space: pre-wrap; overflow-wrap: anywhere; font: inherit; font-size: 12px; line-height: 1.65; max-inline-size: 100%; /* D48 */ }
/* D191 (#542): each node of "All eligible nodes" reviews under its own heading. */
.prompt-sender-node { border-block-start: 1px solid var(--ca-line); padding-block-start: 14px; margin-block: 18px; min-inline-size: 0; /* D191 */ }
.prompt-sender-node > h4 { font-size: 15px; /* D191 */ }
.prompt-sender-field textarea { inline-size: 100%; min-block-size: 100px; resize: vertical; /* D48 */ }
.prompt-sender-composition { padding: 12px; background: var(--ca-field); border-radius: 6px; /* D48 */ }
.prompt-sender-actions { justify-content: flex-end; margin-block-start: 18px; /* D48 */ }
@@ -528,6 +531,10 @@ pre.prompt-text { font: inherit; font-size: 12px; line-height: 1.75; margin-bloc
.instructions-layout { grid-template-columns: 1fr; /* D203 */ }
}
/* D204 (#431): the one-off edit's note under "Tagging instructions" in the Tag image and Auto-tag selected dialogs. */
.tag-model-panel .instructions-edited { margin: -6px 0 12px; /* D204 */ }
.tag-model-panel .instructions-edited[hidden] { display: none; /* D204 */ }
/* D132: at phone width the concept's 120px cap cuts every engine label ("WD14 SwinV2 · ~400 MB"); the Tagging model
select alone keeps its natural width there (its widest engine label) and, with no room beside the row text, wraps
under it. Existing tags (#398) never shrinks for its row text and, past the cap, keeps its natural width beside the