Compare commits

...
Author SHA1 Message Date
Vito Sansevero a83799451b ci(browser): load the running-tests reporter by absolute path
A config imported from another directory (the workflow-artifacts test copies
it to a temp root) resolved './running-tests-reporter.mjs' against that root
under CI and failed to load it. Resolve it from this module instead.
2026-10-28 11:15:17 -07:00
Vito Sansevero d5f642dc4b Merge remote-tracking branch 'origin/main' into ci/browser-hang-diagnostics 2026-10-28 10:47:03 -07:00
Vito Sansevero 216267348d ci(browser): forward SIGTERM and SIGINT from frontend-coverage.mjs to its child
In the coverage job npm signals only frontend-coverage.mjs. It blocked in
spawnSync with no listener, so Playwright never saw a step timeout. It now
runs each step's child with async spawn, in its own process group, and
forwards SIGTERM and SIGINT to the running child. Step order, exit codes
and the "Command failed (N)" message are unchanged.

Locally, in the pinned image, `timeout -s TERM` on test:coverage printed
the still-running tests and wrote both blob reports. TESTING.md records
that a SIGTERM to npm alone stops at npm's `sh -c`, and that delivery
through `docker exec` in CI is not yet observed.

Refs #425
2026-10-28 10:47:03 -07:00
Vito Sansevero f675d4de2b ci(browser): rely on the step limit, report the running tests on SIGTERM
Maintainer ruling on #425: no Playwright globalTimeout. The 18-minute
step limit under the unchanged 20-minute job limit stays, as do the
always() uploads and the CI blob report.

- Drop globalTimeout and PM_TEST_GLOBAL_TIMEOUT_MINUTES with its helper
  and tests.
- A step that reaches its limit gets SIGTERM, which ended Playwright with
  no report and no blob. The running-tests reporter now catches the first
  SIGTERM, prints the tests still running and sends Playwright its own
  SIGINT, so the run ends interrupted and the blob report is written. A
  second SIGTERM kills as usual.
- Test paths are relative to the suite's rootDir instead of the cwd, so
  `npm run --prefix` no longer prints ../../ prefixes.
- The workflow comment and TESTING.md say the guarantee assumes an npm
  cache hit and that coverage shard 1 is tight in its 18-minute step.
2026-10-28 10:37:54 -07:00
Vito cdb771d722 feat(images): filter, inspect and count by each image's effective category (#332, PR 2) (#406)
* feat(images): store images' own categories in the library database (#332)

One new table, image_categories, and its category index, created empty
by a forward-only migration that reads and writes only its own objects.
Startup checks the structure; restore also checks the values. The store
sets, clears and reads own categories by file key (the deepest output
source holding the file and the path below it), and copies a removed
duplicate's category to a kept copy that has none.

* feat(images): duplicate cleanup keeps images' own categories (#332)

The kept copy's own category wins; a kept copy without one receives the
removed copy's, keyed by the deepest output source holding each file.
The removed copy's row is never deleted, so a give-back or a failed
delete loses nothing. The job log reports how many were kept.

* feat(images): set, remove and list images' own categories (#332)

PUT /prompt_manager/images/category sets the own category of up to 500
listed files, DELETE removes it (listed or not), and
GET /prompt_manager/images/categories lists the values whose file is
listed. Targets are the Images listing identity of an output source,
checked by media_target (shared with Collections). The routes never
touch prompts, recipes or links; POST /prompt_manager/bulk/category is
unchanged.

* feat(vocabulary): count and rename image categories (#332)

Tags & categories lists every category value with image_count, the
stored image_categories rows holding it. Values only images hold follow
every prompt-backed value (first_pid at SQLite's INTEGER maximum), so
keyset cursors cross the boundary. /library/options keeps its own query
and never lists image-only values.

Manage category counts image rows (impact.images), treats an image-held
target as existing, renames image rows in the same transaction, allows
renaming an image-only value, and hashes the sorted image rowids into
the impact token so a change between preview and apply is refused with
vocabulary-conflict. Every image-category statement reads
idx_image_categories_category.

* feat(images): the category filter matches each image's effective category (#332)

An image's own category wins; without one it inherits its link's: the
recipe's category when set, else the positive prompt's (MC4, one rule
with the inspector chip). Ambiguous and unlinked images inherit nothing,
so an unlinked image with its own category is now listed. The link read
no longer filters by category in SQL; the page filter, select all N
matching and saved views compare the effective value. A failed
own-category read answers the existing listing-failed 500.

* fix(images): key symlinked output folders and keep key work off the event loop (#332)

Review fixes: a source matches by its configured spelling too, so a
detected output folder that is a symlink keys its files (routes and
duplicate cleanup). File keys are computed in the executor. A path no
source holds is a plain 403, a lone surrogate a 400. The install repairs
a table that lacks its category index instead of refusing to open.

* feat(inspector): show an image's own category in the inspector and History (#332)

The inspector's library facts gain category_scope "image": an image's
own category overrides the inherited one for any link status (linked,
ambiguous, none). Images reads it for output sources only; History reads
it once per inspection inside its snapshot, keyed by the same file key
from the symlink-resolved output folder. A failed read hides the chip
rather than show a possibly wrong value. The chip's hidden prefix names
the origin ("Image category: "); the visible origin text comes in PR 3.

* perf(images): keep the category filter's link read narrow; record D141, D142, D145 (#332)

Under a category the link read keeps only links whose raw recipe or
positive category is the value (Python still decides the inherited
one), the links of images whose own category is the value are read by
path, and plain reads skip the inherited-category work. The own-category
map is keyed by normalized path. TESTING.md records D141, D142 and D145
and amends D13, D36 and D39; the filter budget seeds 2,000 own
categories.

* feat(images): set or remove a category on all N matching images (#332)

The category routes also accept the Images listing filter of "Select
all N matching" (#392) instead of explicit items, resolved now and
capped at 500 images like an explicit selection.

* fix(images): Windows path keys for own categories; plan check for the category read (#332)

Review fixes: on Windows the own-category map keys each file with
normalize_image_path (abspath drops trailing dots and spaces), and the
plan test also covers the category's narrowing clause.

* test(vendor): count the #332 host case in the execution matrix

image-attribution.spec.mjs gains the image category route case, so the
execution config lists 92 tests (3 attribution cases per project).

* test(legacy): record the image category routes in the legacy contract snapshot (#332)

#332 adds PUT/DELETE /prompt_manager/images/category and
GET /prompt_manager/images/categories; the legacy pages' routes are
unchanged.
2026-10-28 17:19:11 +00:00
Vito Sansevero f2042b06bf ci(browser): stop a hung browser run before the job limit and name it
A browser job killed at its 20-minute limit keeps no log and skips its
upload steps (merge-queue run 40723857746, firefox shard 2/2), so nothing
showed which test hung.

- Each browser test step gets timeout-minutes 18, below the job's 20, so
  the step fails and the always() uploads still run.
- Under CI, Playwright's globalTimeout stops the run first:
  PM_TEST_GLOBAL_TIMEOUT_MINUTES 17 for page workflows, 15 for coverage
  workflows, a fixed 3 for the viewer suite. Locally there is no limit
  unless the variable is set. Each test keeps the 30 s default.
- running-tests-reporter.mjs prints the tests still running when the run
  ends; Playwright's own reporters cannot, since those tests never end.
- A blob report goes to tools/vendor/test-results/blob/, inside the
  uploaded traces path.
- ci-workflow.test.mjs pins globalTimeout < step limit < job limit and the
  always() uploads; no existing limit is raised.

Closes #425
2026-10-28 10:07:14 -07:00
Vito 55fa2c4870 fix(runs): count cached PromptManager nodes fed by a runtime text link on every job (#347) (#423)
* fix(usage): count cached re-runs of runtime-linked prompt text (#347)

A PromptManager-family node whose text comes from a runtime link
(PreviewAny, an LLM, a list node) counted only the job that executed it;
later jobs served from ComfyUI's output cache counted nothing.

Each execution now records a receipt (digests of its STRING outputs and
hashes of the text it stores) per run key and queue item. On
execution_cached the receipt whose digests equal the cache entry's text is
replayed once for the job, with the node's role and what the queue hook
already saw recomputed from the job's own graph. The executing node and the
replay count through one shared own_counts. Unmatched cache entries count
nothing, as before; a node that executes after its replay in the same job
skips its own count. No schema change, no backfill.

* fix(usage): never replay a node twice in one job; dedupe receipts once complete (#347)

A repeated execution_cached id or message for the same job now counts the
node once. Receipts replace equal older ones only when their job finishes
(or the next job starts), so a list still recording [A] of [A, B] no longer
drops an older [A] receipt. Node-level tests check the receipt each
execution records.

* test(comfyui): count #347's cached-run cases in the execution suite pin (#347)

The pin listed 19 execution-text-link cases per project; #347 adds five node
kinds and eight one-renderer cases, so the suite lists 32 per project and 118
in total with #332's image-attribution case from main.
2026-10-28 16:37:24 +00:00
Vito 11422d4584 perf(images): read prompt links in one SQLite step (#354) (#418)
* perf(images): read prompt links in one SQLite step (#354)

get_image_link_matches streamed one sqlite3_step per linked row, and
Python's sqlite3 releases and retakes the GIL around every step. While
ComfyUI generates, each retake waits on the busy thread, so the
first-page link read of the workspace Images page stretched from tens of
milliseconds to seconds.

SQLite now returns the unchanged filter SELECT's rows as one JSON value
(json_group_array), decoded with json.loads: a constant two or three GIL
hand-backs instead of one per row. Nothing is cached. Any
OperationalError from the wrapped statement, or a BLOB file_path or
rating, makes that read stream as before with one warning per process; a
missing JSON function stops the attempt for the rest of the process.

tests/test_database_link_matches.py proves the result equals the
streamed read on every filter combination and on NULL ratings, quoted,
backslashed, non-ASCII and multi-line paths, ambiguous and pair links,
an empty database and both fallbacks; the structural budget test also
checks the wrapped query plan.

* fix(images): keep the maintenance link reader, refine the one-step fallback (#354)

The one-step reader was named _link_rows, which shadowed
MaintenanceOperations._link_rows on PromptDatabase and broke the Missing
links preview and apply; it is now _one_step_link_rows, and a test pins
that the maintenance reader is not shadowed.

An interrupted read (Move library retiring the connection) is raised as
before instead of falling back and spending the one warning; a JSON text
over SQLite's length limit (DataError) streams like the other value
failures. The equivalence test compares value types per path rather than
by row order, and a test checks that the read leaves no WAL snapshot
pinned.

* docs(testing): name every one-step link read fallback (#354)
2026-10-28 16:00:15 +00:00
42 changed files with 4218 additions and 209 deletions
+13
View File
@@ -27,6 +27,15 @@ concurrency:
# coverage shards 7m34s/6m04s/5m22s (tests 6m38s/4m53s/4m04s), merge 54s, firefox 7m57s/5m21s (tests
# 7m08s/4m34s), webkit 8m16s/5m03s (tests 7m26s/3m58s); a cache miss adds up to 5m to npm ci, which
# the 20-minute limits absorb.
# Hangs (#425): a job killed at its limit keeps no log and skips its upload steps, so each browser
# test step has its own timeout-minutes (18) below the job's 20. At the step limit, SIGTERM makes
# tools/vendor/running-tests-reporter.mjs print the tests still running and hand over to Playwright's
# interrupt, which writes the blob report; the always() uploads then run.
# Measured on main (runs 40725289094, 40719375442, 40700439399): setup before the test step <= 52s,
# uploads <= 35s; test steps up to 14m42s (webkit shard 1/2) and 13m05s for coverage (shard 1/3:
# 12m53s, node units and the viewer suite before 11.8m of workflows), so coverage is the tight one
# in its 18 minutes. This holds only when the npm cache hits: a cold cache adds up to 5m to npm ci,
# and then the job limit can still end the job before the step limit does.
jobs:
frontend-coverage:
name: frontend coverage (shard ${{ matrix.shard }}/3)
@@ -79,6 +88,7 @@ jobs:
- if: matrix.shard == 3
run: npm run --prefix tools/vendor test
- name: Page workflows with coverage, shard ${{ matrix.shard }}/3
timeout-minutes: 18
run: npm run --prefix tools/vendor test:coverage:shard -- ${{ matrix.shard }}/3
- name: Upload raw coverage
uses: actions/upload-artifact@v7
@@ -87,6 +97,7 @@ jobs:
path: coverage/js/raw/
if-no-files-found: error
retention-days: 3
# Traces and the blob report (test-results/blob/) of a failed, timed-out or cancelled step.
- name: Upload Chromium and touch traces
if: always()
uses: actions/upload-artifact@v7
@@ -211,7 +222,9 @@ jobs:
key: npm-${{ runner.os }}-playwright-v1.64.1-${{ hashFiles('tools/vendor/package-lock.json') }}
- run: npm ci --prefix tools/vendor --ignore-scripts
- name: Page workflows in ${{ matrix.browser }}, shard ${{ matrix.shard }}/2
timeout-minutes: 18
run: npm run --prefix tools/vendor test:workflows -- --project ${{ matrix.browser }} --shard ${{ matrix.shard }}/2
# Traces and the blob report (test-results/blob/) of a failed, timed-out or cancelled step.
- name: Upload browser traces
if: always()
uses: actions/upload-artifact@v7
+126 -8
View File
@@ -262,6 +262,31 @@ watchdog dump, is a failure like any other: read the dump, find the blocking
call, fix it. Never raise a limit or rerun to get green; a limit changes only
with new measurements written next to it.
Browser jobs (#425): GitHub keeps no log of a job killed at its limit and skips
its upload steps, so the test step in each browser job in `viewerjs-check.yml`
has its own `timeout-minutes` (18, below the job's 20). At the step limit
GitHub sends SIGTERM, which would end Playwright without any report; under `CI`
`tools/vendor/running-tests-reporter.mjs` catches the first one, prints
`Still running when the run ended (SIGTERM): N` and one line per unfinished
test (path relative to the suite's `rootDir`), then sends Playwright its own
SIGINT, so the workers stop, the run ends interrupted and the blob report is
written to `tools/vendor/test-results/blob/`. In the coverage job npm signals
only `frontend-coverage.mjs`, which forwards SIGTERM and SIGINT to the child it
is running (started in its own process group, so a group-wide signal still
arrives once). Delivery is proven locally in the pinned image for a SIGTERM to
the whole process group (`timeout -s TERM`). A SIGTERM to npm alone
(`timeout --foreground`) reaches nothing: npm forwards it to the `sh -c` that
runs the script, and that shell exits without passing it on. Delivery in CI,
where the step runs through `docker exec`, is not yet observed; the next real
step timeout will show it. That directory and the traces
upload with `if: always()`; `npx playwright merge-reports` opens the blob.
Each test keeps the 30 s default and there is no `globalTimeout`. Two limits
of this: it assumes the npm cache hits (a cold cache adds up to 5 minutes to
`npm ci`, and the job limit can then still win), and coverage shard 1 (12m53s
measured) is the tight one in its 18-minute step. Local runs keep the `list`
reporter only; reproduce with `CI=1 timeout -s TERM 40 npm run --prefix
tools/vendor test:workflows -- --project=firefox`.
### After a push
Follow the run for the commit you pushed, never an older one:
@@ -391,6 +416,10 @@ A spec performance budget gets two tests:
`test_filter_budget` (Images filters, spec D12: p95 < 150 ms at 50,000 files
and 20,000 links).
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`).
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
@@ -1340,7 +1369,7 @@ How the earlier workspace Images filters moved (old → new). The legacy `/galle
| Folder | "Refine this view" → Folder (D14); shown in the scope chip |
| Media (All media / Images / Videos) | "Refine this view" → Media (D14) |
| Sort: Newest first / Oldest first / Filename A–Z / Filename Z–A | toolbar sort: Recently added / Oldest first / Name A–Z / Name Z–A, plus Highest rated (D13, D14) |
| (none) | toolbar category and "Refine this view" → Tags, through the linked positive (D13, D14) |
| (none) | toolbar category (the effective category, D141) and "Refine this view" → Tags, through the linked positive (D13, D14) |
| Refresh | Import → "Scan an output folder" → Rescan (D15) |
| Grid / List / Masonry, Comfortable / Large / Compact | the same choices in the concept filter row |
| Card image → side panel; preview icon → viewer | image → viewer; Info → side panel (D19) |
@@ -1423,7 +1452,8 @@ Recorded deviations on this page (spec §8):
`overscroll-behavior: contain`.
- **D35**: counts pluralize ("1 positive prompt", "1 image", "1 more image loaded").
- **D36**: the category list is the Library's, sorted alphabetically (case-insensitive)
by the page, whatever order the server sends.
by the page, whatever order the server sends (amended by D141: Images adds its
image categories).
- **D37**: production loads 40 images per page (the concept 12), so the continuation
and toast use 40; after Escape production returns focus to the opening image.
- **D44** (#249 PR E): Filters reads "Filters · N" while N of Source, Folder, Media and
@@ -1509,15 +1539,46 @@ Recorded deviations on this page (spec §8):
- **D39** (ruled 2026-10-21: fallback): a recipe-linked image shows its recipe's own
category when set, otherwise the linked positive prompt's, and its recipe's own tags
when it has any, otherwise the positive's; each field falls back on its own and the
two tag sets are never merged. Set category (#259) writes the positive's category and
the Images category and Tags filters match the positive's, so the panel now agrees
with them whenever the recipe adds nothing. The inspection response names each
two tag sets are never merged. The Images Tags filter matches the positive's tags.
The category filter matches the same inherited category as the panel (D142, #332), and
an image's own category overrides both (D141, D146). The inspection response names each
field's source (`library.category_scope`, `library.tags_scope`: `pair`, `positive` or
null); the category chip's visually hidden prefix (D21) says "Recipe category: " or
"Positive prompt category: " from it, and tag chips are Tags filter buttons only when
they are the positive's (#260 D2). History's media inspector uses the same rule. The
concept shows the prompt's category and tags for every image and the concept
fixture's recipes have neither, so the inspector comparison needs no mapping for it.
- **D141** (#332, amends D13 and D36): the Images category filter matches each image's
effective category: its own category (Images Set category, PR 3) when it has one,
else its inherited one (D142), else none. An image whose own category differs never
matches its prompt's category, and an unlinked or ambiguous image with an own
category is listed under it (the members-only gate now applies to the Tags filter
only). "Select all N matching" and saved views use the same value. The caption's
positive-prompt count still counts the linked positives of the images shown. With a
category set the listing reads the own categories once per source (a primary-key
range); a failed read is the existing 500 `listing-failed`, never an unfiltered page.
The link read then keeps only links whose raw recipe or positive category is the
value (a superset of those that inherit it; Python decides), and reads the links of
files whose own category is the value by path, so their rating and positive count.
`test_filter_budget` seeds 2,000 own categories on its 50,000-file fixture.
The option list adds the image categories of listed files
(`GET /prompt_manager/images/categories`, PR 3).
- **D142** (#332 MC4, amends D39's filter note): the inherited category is one rule,
`inherited_category` in `utils/inspection_response.py`: a linked recipe's own
category when it is not blank, else the linked positive's. Before #332 the filter
matched the positive's even when the chip showed the recipe's. The link reads select
the raw recipe and positive categories and never decide inheritance in SQL
(`tests/test_api_workspace_image_categories.py` checks the filter and the chip agree
on blank values stored as spaces, tabs, U+00A0 and BLOB). The concept has no recipe
categories, so this is behavior only.
- **D145** (#332, DC4): History's media inspector shows an image's own category with
the chip rule of the side panel and viewer. Only the inspection request reads it, once,
inside the History snapshot: the file is the symlink-resolved output folder plus the
receipt's subfolder and filename, keyed with `image_category_key` over the configured
output sources, as Images keys it. A temp output or a file no source holds has none;
a failed read hides the chip (DC5). The outputs list and media requests read nothing
new. The chip's hidden prefix reads "Image category: " for an own category; the
visible origin text and its comparison are D146 (PR 3).
- **D42**: the concept never shows a toast and a page alert together. Production keeps
the concept's fixed toast (centred, 22px from the bottom, 12px at 700px and below)
and, when an Import → Rescan toast leaves the Images discovery alert
@@ -4624,8 +4685,18 @@ Filters (`mode=workspace`; anything invalid is 400 "Invalid gallery filters"):
rating 5 to 1, then linked but unrated, then unlinked and ambiguous; ties break
newest first. A signed cursor covers every filter and the sort, so reusing it
with a change is 400.
- One link read (`get_image_link_matches`, streamed and bounded to the listed
paths) serves the filters, the rating sort and the count.
- One link read (`get_image_link_matches`) serves the filters, the rating sort
and the count. SQLite returns its rows as one JSON value in one step (#354):
Python's `sqlite3` releases and retakes the GIL around every step, and while
ComfyUI generates each retake can wait on the busy thread, so a read that
stepped once per linked row took seconds. The result keeps only the listed
paths; peak memory grows with the database's matching linked rows (about
530 bytes each, freed on return). When the JSON step fails (a SQLite without
JSON functions, a BLOB value, a JSON text over SQLite's length limit) the read
streams the rows as before and logs one warning per process; a missing JSON
function stops the attempt for the rest of the process. An interrupted read
(Move library) raises as before. `tests/test_database_link_matches.py` proves both reads
equal on every filter and on odd values.
`positive_count` is the number of distinct linked positives over the full matching
set, not the page:
@@ -5269,7 +5340,7 @@ which used to save `null` under a link.
Primitive or through `PreviewAny`, for PromptManager, PromptManagerText, Focused, Split
and the Pair (positive, negative, both, neither linked, and a Primitive negative). A
wildcard Primitive on PromptManager (queued as written) and Focused (one choice) saves
and counts what was queued. The linked text is saved, counted once, recorded as
and counts what was queued. The linked text is saved, counted once per job, recorded as
the run's raw positive/negative, and after a scan the run's image links to it. Text
only known at run time (`PreviewAny`) cannot be read from the image; the case only
requires that it is never linked to another prompt (its attribution from the recorded
@@ -5283,6 +5354,39 @@ PM_COMFY_ROOT=... PM_TEST_PYTHON=... npm run --prefix tools/vendor test:comfyui
PM_COMFY_ROOT=... PM_TEST_PYTHON=... npm run --prefix tools/vendor test:executions -- execution-text-link
```
## Cached runtime-linked runs (#347)
Text only known at run time (a `PreviewAny` or LLM link, a list node) is counted by the
node when it executes; the queue hook cannot read it. When ComfyUI serves such a node from
its output cache, `utils/cached_runs.py` counts the job instead: every execution records
a receipt (digests of its STRING outputs and hashes of the text it stores, per run key and
queue item), and on `execution_cached` the receipt whose digests equal the cache entry's
text is replayed once. The role (positive, negative, which Pair outputs are used) and what
the hook already saw are recomputed from the job's own graph, and the node and replay
decide through the same `own_counts`.
- Fail closed: no receipt in this process (after a restart, an external cache provider),
changed cached text, more than 64 list values, a node the job's outputs do not need, or
a host without the queue/executor interfaces: nothing is counted, as before #347.
- A cached node counts at job start like the hook: a job failing later still counts it.
A node that executes after its replay in the same job (RAM-pressure freed the entry)
skips its own count but still consumes the pending first-use catch-up.
- No backfill: earlier cached jobs keep their counts. The history page's "tracked runs"
can stay below the Library count for these prompts (#403).
- Tests: `tests/test_cached_runs.py` (whole jobs with the real nodes, a real database,
fake host caches); `tests/comfyui/execution-text-link.spec.mjs` "#347" cases on the
managed host: PreviewAny into each node kind, 3 jobs changing only the sampler seed,
count 3 in both renderers; in one renderer the Classic, LRU, RAM-pressure and no-cache
modes, A, B, A from the owned `PMCycledText` under LRU, typed text, a downstream
failure, a "run selected output" job that does not need the node, and a rewire to the
negative input. The RAM-pressure eviction path cannot be forced on a real host and is
covered by the Python tests.
```sh
python3 -m unittest discover -s tests -p "test_cached_runs.py"
PM_COMFY_ROOT=... PM_TEST_PYTHON=... npm run --prefix tools/vendor test:executions -- execution-text-link --grep "#347"
```
## Durable execution ledger
`npm run --prefix tools/vendor test:executions` runs 52 real execution cases
@@ -5702,6 +5806,20 @@ rotates the epoch, so a pre-restore `expected_epoch` is refused with `storage-ch
Owner rows and their timestamps never change on a tag command; a category apply only
touches rows carrying the exact source.
Image categories (#332 §7, MC9, MC10, DC7): each `GET categories` row also carries
`image_count`, the stored `image_categories` rows holding that exact value (files
missing on disk included). A value prompts and images share appears once, ordered by
its oldest positive; a value only images hold appears once after every prompt-backed
value (`first_pid` 9223372036854775807 in the keyset cursor), ordered by value, with
`prompt_count` 0. `/library/options` keeps its own query and never lists image-only
values. Category preview and apply report `impact.images` (rows holding the source),
the target exists when an image row holds it, an image-only value can be renamed, the
apply renames the image rows in the same transaction (file keys and rowids kept, so a
merge cannot collide), and the impact token hashes the sorted image rowids, so an image
row gaining or losing the value between preview and apply is 409
`vocabulary-conflict` with nothing written. Every image-category statement reads
`idx_image_categories_category` (`test_category_reads_use_the_image_category_index`).
Two legacy routes change additively: `GET /tags` now lists unused registry tags, and
`/tags/suggest` without context appends unused names at count 0. Unused tags become
deletable only through the native Delete tag (vocabulary UI PR), not on the legacy
+20
View File
@@ -195,6 +195,26 @@ except Exception as e:
f"Failed to register usage hook: {e}"
)
# A cached node counts what its last execution of the same text counted (#347).
try:
from .py.config import server_instance
from .utils.cached_runs import register_cached_runs
register_cached_runs(server_instance, _db_factory)
except Exception as e:
try:
from .utils.logging_config import get_logger
get_logger("prompt_manager.init").error(
f"Failed to register cached-run counting: {e}"
)
except Exception:
import logging
logging.getLogger("prompt_manager.init").error(
f"Failed to register cached-run counting: {e}"
)
def _history_db_factory() -> "PromptDatabase":
"""Initialize history on its worker even if monitoring initialization failed."""
+184 -41
View File
@@ -7,7 +7,12 @@ import json
import math
import datetime
import os
from typing import Optional, List, Dict, Any, Set, Tuple
from typing import Any, Dict, Iterable, List, Optional, Sequence, Set, Tuple
try:
from ..utils.inspection_response import inherited_category
except ImportError:
from utils.inspection_response import inherited_category
from .models import PromptModel, normalize_image_path, utc_now_iso
from .pairs import PairDependencyError, PairOperations # noqa: F401 (re-exported)
@@ -238,12 +243,17 @@ LEFT JOIN prompt_library_state pls ON pls.prompt_id = pp.positive_prompt_id
WHERE gi.image_path IN ({marks}) OR gi.file_path IN ({marks}){fallback}"""
# One image row describes a card only when no other row claims the same file
# (as in get_image_link_summaries, several rows are ambiguous). The pair scope
# and the linked positive follow _LINK_SUMMARY_SQL; category, rating, tags and
# the searchable text and titles come from that positive (spec D12/D13). Every
# and the linked positive follow _LINK_SUMMARY_SQL; rating, tags and the
# searchable text and titles come from that positive (spec D12/D13). The raw
# recipe and positive categories are read for the inherited category, which
# Python decides (``inherited_category``, #332 MC4); SQL never filters by
# category. Every
# writer since the 3.2.4 migration stores file_path, so it is the path key.
# The prompt-level conditions are uncorrelated IN sets SQLite builds once, and
# the CASE evaluates the per-row ambiguity probe only for admitted rows.
_LINK_FILTER_SQL = """SELECT gi.file_path, p.rating, {hit}, p.id
_LINK_FILTER_SQL = """SELECT gi.file_path AS file_path, p.rating AS rating,
{hit} AS matched, p.id AS positive, pp.id IS NOT NULL AS has_pair,
pp.category AS pair_category, p.category AS category
FROM generated_images gi
LEFT JOIN prompt_pairs pp ON pp.id = (
SELECT CASE WHEN COUNT(DISTINCT ip.pair_id) = 1 THEN MIN(ip.pair_id) END
@@ -252,10 +262,31 @@ JOIN prompts p ON p.id = COALESCE(pp.positive_prompt_id, gi.prompt_id)
WHERE gi.file_path IS NOT NULL AND CASE WHEN {admit} THEN NOT EXISTS (
SELECT 1 FROM generated_images other
WHERE other.file_path = gi.file_path AND other.id <> gi.id) ELSE 0 END"""
# _LINK_FILTER_SQL's rows as one JSON array, built by SQLite in one step (#354).
# Python's sqlite3 releases and retakes the GIL around every step, so a read
# that steps once per linked row waits on a busy ComfyUI thread thousands of
# times. A BLOB value (no writer stores one) makes the value NULL (or
# json_array refuses it), and the read streams instead. The raw categories are
# put in only for a category read (#332), so plain reads stay as small as #354
# measured them.
_LINK_MATCHES_JSON_SQL = """SELECT CASE
WHEN max({blob}) THEN NULL
ELSE json_group_array(json_array({columns})) END
FROM ({select})"""
_LINK_PLAIN_COLUMNS = ("file_path", "rating", "matched", "positive")
_LINK_CATEGORY_COLUMNS = _LINK_PLAIN_COLUMNS + ("has_pair", "pair_category", "category")
# The columns a BLOB can reach: matched, positive and has_pair are computed.
_LINK_STORED_COLUMNS = ("file_path", "rating", "pair_category", "category")
# Set when this SQLite has no JSON functions: every later read streams.
_one_step_unsupported = False
# One warning per process when a read falls back to streaming.
_one_step_warned = False
# The linked positive of each given path (one unambiguous link, pair scope as
# in _LINK_FILTER_SQL) with its category; {hit} as in link_filter_query. Used
# for the few files whose current image tags meet a Tags filter (#249 B3).
_LINK_FACTS_SQL = """SELECT gi.file_path, p.rating, {hit}, p.id, p.category
# in _LINK_FILTER_SQL) with the raw categories of its inherited category; {hit}
# as in link_filter_query. Used for the few files whose current image tags meet
# a Tags filter (#249 B3).
_LINK_FACTS_SQL = """SELECT gi.file_path, p.rating, {hit}, p.id,
pp.id IS NOT NULL, pp.category, p.category
FROM generated_images gi
LEFT JOIN prompt_pairs pp ON pp.id = (
SELECT CASE WHEN COUNT(DISTINCT ip.pair_id) = 1 THEN MIN(ip.pair_id) END
@@ -284,8 +315,24 @@ _LINK_FILTER_TAG = (
)
# One unambiguously linked image's facts for the Images filters, a plain tuple
# (hit, rating, positive id, inherited category): built per linked row, so no
# NamedTuple. The category is the inherited one (#332 MC4: the recipe's when
# set, else the positive's), never the image's own category.
LinkMatch = Tuple[bool, Optional[int], int, Any]
def _link_match(rating, matched, positive, has_pair, pair_category, category):
inherited, _scope = inherited_category(category, pair_category, bool(has_pair))
return (bool(matched), rating, positive, inherited)
def link_filter_query(
q: str, category: Optional[str], tags: List[str], *, every_link: bool = False
q: str,
tags: List[str],
*,
every_link: bool = False,
category: Optional[str] = None,
) -> Tuple[str, Dict[str, Any]]:
"""The workspace filter SELECT and its named parameters (all bound).
@@ -299,7 +346,9 @@ def link_filter_query(
params["pattern"] = f"%{escape_like(q)}%"
admitted = []
if category is not None:
admitted.append("p.category = :category")
# Only narrows: an inherited category is one of the two raw values;
# which one is decided in Python (inherited_category, #332 MC4).
admitted.append("(p.category = :category OR pp.category = :category)")
params["category"] = category
for index, tag in enumerate(tags):
admitted.append(_LINK_FILTER_TAG.format(index=index))
@@ -310,6 +359,50 @@ def link_filter_query(
return sql, params
def link_matches_json_query(sql: str, categories: bool = False) -> str:
"""*sql* from link_filter_query, wrapped to return its rows as one JSON value.
With *categories* each row also carries the raw values the inherited
category is decided from; otherwise only path, rating, hit and positive.
"""
columns = _LINK_CATEGORY_COLUMNS if categories else _LINK_PLAIN_COLUMNS
blob = " OR ".join(
f"typeof({name}) = 'blob'" for name in _LINK_STORED_COLUMNS if name in columns
)
return _LINK_MATCHES_JSON_SQL.format(
blob=blob, columns=", ".join(columns), select=sql
)
def _link_matches_from_rows(
rows: Iterable[Sequence[Any]], categories: bool, wanted: Optional[Set[str]]
) -> Dict[str, LinkMatch]:
"""LinkMatch per wanted path from link rows (one-step or streamed).
Without *categories* the inherited category is None, as the caller then
never compares it.
"""
matches: Dict[str, LinkMatch] = {}
for path, rating, matched, positive, *raw in rows:
if wanted is None or path in wanted:
if categories:
matches[path] = _link_match(rating, matched, positive, *raw[:3])
else:
matches[path] = (bool(matched), rating, positive, None)
return matches
def _streamed_link_matches(
conn: sqlite3.Connection,
sql: str,
params: Dict[str, Any],
wanted: Optional[Set[str]],
categories: bool = False,
) -> Dict[str, LinkMatch]:
"""The link read stepped once per row (the fallback of the one-step read)."""
return _link_matches_from_rows(conn.execute(sql, params), categories, wanted)
_ASCII_FOLD = str.maketrans("ABCDEFGHIJKLMNOPQRSTUVWXYZ", "abcdefghijklmnopqrstuvwxyz")
@@ -2543,38 +2636,94 @@ class PromptDatabase(
def get_image_link_matches(
self,
q: str,
category: Optional[str],
tags: List[str],
*,
every_link: bool = False,
wanted: Optional[Set[str]] = None,
) -> Dict[str, Tuple[bool, Optional[int], int]]:
"""Map normalized paths of unambiguously linked images to
(hit, rating, positive id).
category: Optional[str] = None,
) -> Dict[str, LinkMatch]:
"""Map normalized paths of unambiguously linked images to a LinkMatch
(hit, rating, positive id, inherited category).
Rows are limited to images whose linked positive has *category* and
every tag in *tags* (exact names). ``hit`` says whether *q* occurs in
Rows are limited to images whose linked positive has every tag in
*tags* (exact names). The category filter is applied by the caller to
the effective category (#332). ``hit`` says whether *q* occurs in
the positive's text, Library title or a tag name; a pair's own title
names its negative variant and is not searched.
Without category or tags only hits are returned, unless *every_link*
Without tags only hits are returned, unless *every_link*
asks for every linked image (the rating sort and the positive count).
The positive id is the card's linked positive. One SELECT whose rows
are streamed; with *wanted* (the listing's normalized paths) only those
paths are kept, so memory is bounded by the listing, not the database.
The positive id is the card's linked positive. With *category* only
links whose raw recipe or positive category is that value are read (a
superset of the links that inherit it; the caller compares the
inherited value) and the inherited category is decided per row;
otherwise it is None, which keeps the plain reads cheap.
One SELECT, returned as one JSON value (#354): SQLite steps once, so
the read takes the GIL back a constant two or three times instead of
once per linked row while ComfyUI generates. Peak memory therefore
grows with the database's matching linked rows (about 530 bytes each,
more with the raw categories of a category read, freed on return); the
result is bounded by *wanted* (the listing's normalized paths) when
given. Where the JSON step fails (no JSON functions, a BLOB value, a
JSON text over SQLite's length limit) the rows are streamed as before.
"""
sql, params = link_filter_query(q, category, tags, every_link=every_link)
matches: Dict[str, Tuple[bool, Optional[int], int]] = {}
sql, params = link_filter_query(
q, tags, every_link=every_link, category=category
)
categories = category is not None
with self.model.get_connection() as conn:
for path, rating, matched, positive in conn.execute(sql, params):
if wanted is None or path in wanted:
matches[path] = (bool(matched), rating, positive)
return matches
rows = None
if not _one_step_unsupported:
rows = self._one_step_link_rows(conn, sql, params, categories)
if rows is None:
return _streamed_link_matches(conn, sql, params, wanted, categories)
return _link_matches_from_rows(rows, categories, wanted)
def _one_step_link_rows(
self,
conn: sqlite3.Connection,
sql: str,
params: Dict[str, Any],
categories: bool = False,
) -> Optional[List[List[Any]]]:
"""*sql*'s rows from one JSON value, or None when this read must stream.
An interrupted read (Move library retiring the connection) is raised
as before, never retried row by row.
"""
global _one_step_unsupported, _one_step_warned
error: Optional[str] = None
try:
cursor = conn.execute(link_matches_json_query(sql, categories), params)
try:
((document,),) = cursor.fetchall()
finally:
cursor.close() # resets the statement: its snapshot ends now
except (sqlite3.OperationalError, sqlite3.DataError) as failure:
# DataError: the JSON text is longer than SQLITE_LIMIT_LENGTH.
error = str(failure)
if error == "interrupted":
raise
if error.startswith("no such function"):
_one_step_unsupported = True
else:
if document is not None:
return json.loads(document)
error = "a BLOB value"
if not _one_step_warned:
_one_step_warned = True
self.logger.warning(
"Reading image links row by row (slower while ComfyUI "
"generates); the one-step read failed: %s",
error,
)
return None
def get_image_link_facts(
self, paths: List[str], q: str = ""
) -> Dict[str, Tuple[bool, Optional[int], int, Optional[str], Tuple[str, ...]]]:
) -> Dict[str, Tuple[bool, Optional[int], int, Any, Tuple[str, ...]]]:
"""Map each normalized path with one unambiguous link to its positive's
(hit, rating, positive id, category, tag names).
(hit, rating, positive id, inherited category, tag names).
The link is resolved as in get_image_link_matches, and ``hit`` says
whether *q* occurs in the positive's text, Library title or a tag
@@ -2585,7 +2734,7 @@ class PromptDatabase(
if q:
hit = _LINK_FILTER_HIT
params["pattern"] = f"%{escape_like(q)}%"
rows: Dict[str, Tuple[bool, Optional[int], int, Optional[str]]] = {}
rows: Dict[str, LinkMatch] = {}
names: Dict[int, List[str]] = {}
unique = list(dict.fromkeys(paths))
with self.model.get_connection() as conn:
@@ -2594,10 +2743,8 @@ class PromptDatabase(
bound = {f"path{i}": path for i, path in enumerate(chunk)}
marks = ",".join(f":{key}" for key in bound)
sql = _LINK_FACTS_SQL.format(hit=hit, marks=marks)
for path, rating, matched, positive, category in conn.execute(
sql, {**params, **bound}
):
rows[path] = (bool(matched), rating, positive, category)
for path, *row in conn.execute(sql, {**params, **bound}):
rows[path] = _link_match(*row)
positives = sorted({row[2] for row in rows.values()})
for start in range(0, len(positives), _FACTS_CHUNK):
chunk = positives[start : start + _FACTS_CHUNK]
@@ -2610,20 +2757,16 @@ class PromptDatabase(
path: (*row, tuple(names.get(row[2], ()))) for path, row in rows.items()
}
def get_image_paths_matching(
self, q: str, category: Optional[str], tags: List[str]
) -> Set[str]:
"""Normalized paths whose linked positive matches *q*, *category* and *tags*.
def get_image_paths_matching(self, q: str, tags: List[str]) -> Set[str]:
"""Normalized paths whose linked positive matches *q* and *tags*.
See get_image_link_matches; an empty *q* matches every linked image
the category and tags admit.
the tags admit.
"""
return {
path
for path, (hit, _rating, _positive) in self.get_image_link_matches(
q, category, tags
).items()
if hit or not q
for path, match in self.get_image_link_matches(q, tags).items()
if match[0] or not q
}
def get_prompt_id_for_image(self, image_path: str) -> Optional[int]:
+55 -15
View File
@@ -13,7 +13,9 @@ same connection before commit. Owner rows and their timestamps never change.
Category consolidation is previewed in one ``library_snapshot`` and applied in
one ``_pair_transaction`` that recomputes the owner projection and requires the
preview's ``impact_token``; only rows carrying the exact source change.
preview's ``impact_token``; only rows carrying the exact source change. Image
rows holding the source (``image_categories``, #332) are owners too: they are
counted, hashed into the token and renamed with the prompts and pairs.
"""
import sqlite3
@@ -97,16 +99,30 @@ def _validate_tag_id(tag_id: object) -> None:
raise ValueError("Invalid tag ID")
# The /library/options universe: nonblank values on eligible parents or on
# active pairs, grouped exactly (whitespace and case variants stay distinct).
_CATEGORY_GROUPS = (
# Category values in Tags & categories (#332 §7): the prompt-backed values
# (nonblank values on eligible parents or on active pairs, grouped exactly, so
# whitespace and case variants stay distinct; /library/options lists the same
# prompt-backed set with its own query) plus the values only image rows hold.
# A prompt-backed value is ordered by its oldest owning positive; an image-only
# value sorts after all of them (first_pid is SQLite's INTEGER maximum, above
# any prompt id), then by value. ``image_count`` counts stored rows, including
# files missing on disk. No FULL OUTER JOIN, so it runs on SQLite 3.31+.
_IMAGE_ONLY_FIRST_PID = 2**63 - 1
_CATEGORY_ROWS = (
"WITH owners(value,pid) AS ("
"SELECT p.category,p.id FROM prompts p WHERE p.category IS NOT NULL "
"AND TRIM(p.category)!='' AND " + _ELIGIBLE + " "
"UNION SELECT pp.category,pp.positive_prompt_id FROM prompt_pairs pp "
"WHERE pp.archived=0 AND pp.category IS NOT NULL AND TRIM(pp.category)!=''),"
"grouped AS (SELECT value,MIN(pid) AS first_pid,"
"COUNT(DISTINCT pid) AS prompt_count FROM owners GROUP BY value) "
"image_values(value,image_count) AS ("
"SELECT category,COUNT(*) FROM image_categories GROUP BY category),"
"grouped(value,first_pid,prompt_count) AS ("
"SELECT value,MIN(pid),COUNT(DISTINCT pid) FROM owners GROUP BY value "
"UNION ALL SELECT value," + str(_IMAGE_ONLY_FIRST_PID) + ",0 FROM image_values "
"WHERE value NOT IN (SELECT value FROM owners)),"
"counted AS (SELECT g.value,g.first_pid,g.prompt_count,"
"COALESCE(i.image_count,0) AS image_count FROM grouped g "
"LEFT JOIN image_values i ON i.value=g.value) "
)
@@ -117,15 +133,23 @@ _CATEGORY_COMPONENTS = (
"WHERE p.category=? ORDER BY p.id"
)
_CATEGORY_PAIRS = "SELECT id,archived FROM prompt_pairs WHERE category=? ORDER BY id"
# Image rows holding one exact value (#332 DC7), an index search.
_IMAGE_CATEGORY_ROWIDS = (
"SELECT rowid FROM image_categories WHERE category=? ORDER BY rowid"
)
_CATEGORY_EXISTS = (
"SELECT EXISTS(SELECT 1 FROM prompts WHERE category=:v)"
" OR EXISTS(SELECT 1 FROM prompt_pairs WHERE category=:v)"
" OR EXISTS(SELECT 1 FROM image_categories WHERE category=:v)"
)
# Pairs after components, so a failing pair update rolls both back.
# Pairs after components and image rows last, so a failing update rolls every
# earlier one back. An image row is keyed by its file, never by its value, so
# merging into a value other files already hold cannot conflict.
_CATEGORY_UPDATES = (
"UPDATE prompts SET category=?,updated_at=? WHERE category=?",
"UPDATE prompt_pairs SET category=?,updated_at=? WHERE category=?",
"UPDATE image_categories SET category=?,updated_at=? WHERE category=?",
)
@@ -136,8 +160,15 @@ def category_impact_token(
target_exists: bool,
components: list,
pairs: list,
*,
images: list = (),
) -> str:
"""Hash a category projection like ``library_edit_token`` (no secret)."""
"""Hash a category projection like ``library_edit_token`` (no secret).
``images`` are the rowids of the image rows holding ``source`` (#332 DC7).
With none the preimage has no ``images`` key, so a prompt-only token keeps
the value it had before image categories existed.
"""
preimage = {
"epoch": epoch,
"source": source,
@@ -146,6 +177,8 @@ def category_impact_token(
"components": sorted([int(i), bool(flag)] for i, flag in components),
"pairs": sorted([int(i), bool(flag)] for i, flag in pairs),
}
if images:
preimage["images"] = sorted(int(i) for i in images)
return canonical_token(_CATEGORY_DOMAIN, preimage)
@@ -325,7 +358,8 @@ class VocabularyOperations:
pairs = [
[row[0], bool(row[1])] for row in conn.execute(_CATEGORY_PAIRS, (source,))
]
if not components and not pairs:
images = [row[0] for row in conn.execute(_IMAGE_CATEGORY_ROWIDS, (source,))]
if not components and not pairs and not images:
raise LibraryCommandError("vocabulary-conflict", epoch)
target_exists = bool(
conn.execute(_CATEGORY_EXISTS, {"v": target}).fetchone()[0]
@@ -339,9 +373,10 @@ class VocabularyOperations:
"other_prompts": len(components) - eligible,
"negative_variants": len(pairs) - archived,
"archived_negative_variants": archived,
"images": len(images),
},
"impact_token": category_impact_token(
epoch, source, target, target_exists, components, pairs
epoch, source, target, target_exists, components, pairs, images=images
),
}
@@ -365,7 +400,8 @@ class VocabularyOperations:
def apply_vocabulary_category(
self, *, expected_epoch: str, source: str, target: str, impact_token: str
) -> dict:
"""Set the exact ``source`` to ``target`` on every component and pair.
"""Set the exact ``source`` to ``target`` on every component, pair
and image row.
The recomputed projection must still match the preview's token; a
mismatch is ``vocabulary-conflict`` with nothing written.
@@ -422,15 +458,15 @@ class VocabularyOperations:
_limit(limit)
key = _category_after(after)
total = conn.execute(
_CATEGORY_GROUPS + "SELECT COUNT(*) FROM grouped"
_CATEGORY_ROWS + "SELECT COUNT(*) FROM counted"
).fetchone()[0]
where, params = "", []
if key is not None:
where = " WHERE first_pid>? OR (first_pid=? AND value>?)"
params = [key[0], key[0], key[1]]
rows = conn.execute(
_CATEGORY_GROUPS
+ "SELECT value,first_pid,prompt_count FROM grouped"
_CATEGORY_ROWS
+ "SELECT value,first_pid,prompt_count,image_count FROM counted"
+ where
+ " ORDER BY first_pid,value LIMIT ?",
[*params, limit + 1],
@@ -440,7 +476,11 @@ class VocabularyOperations:
return {
"library_epoch": epoch,
"categories": [
{"value": row["value"], "prompt_count": row["prompt_count"]}
{
"value": row["value"],
"prompt_count": row["prompt_count"],
"image_count": row["image_count"],
}
for row in rows
],
"total": total,
+1
View File
@@ -268,6 +268,7 @@ class PromptManager(PromptManagerBase, ComfyNodeABC):
record_runtime_context(
type(self).__name__, text, prepend_text, append_text, encoding_text
)
self._record_cached_item(unique_id, (encoding_text,), storage_text)
return result
@classmethod
+58 -3
View File
@@ -26,6 +26,13 @@ try:
from .utils.prompt_graph import resolve_text, run_prompt_nodes
from .utils.run_keys import executing_run_id, prompt_node_key
from .utils.usage_tracking import PENDING_FIRST_USE
from .utils.cached_runs import (
CACHED_RUNS,
Role,
Seen,
own_counts,
stored_facts,
)
except ImportError:
import sys
@@ -38,6 +45,13 @@ except ImportError:
from utils.prompt_graph import resolve_text, run_prompt_nodes
from utils.run_keys import executing_run_id, prompt_node_key
from utils.usage_tracking import PENDING_FIRST_USE
from utils.cached_runs import (
CACHED_RUNS,
Role,
Seen,
own_counts,
stored_facts,
)
class PromptManagerBase:
@@ -71,8 +85,10 @@ class PromptManagerBase:
"""Save the prompt to the SQLite database.
Runs are normally counted by the queue hook (utils/usage_tracking.py), so a
save only counts one when asked to (batch items whose text the hook could not
see) or when the hook queued this prompt before it existed.
save only counts one when asked to (text the hook could not see, unless
this job already counted the node from ComfyUI's cache) or when the hook
queued this prompt before it existed. The two are independent: the
pending catch-up is written even when ``count_run`` is False.
Args:
text: The prompt text
@@ -167,8 +183,19 @@ class PromptManagerBase:
batch_item = (
role["known"] and role["is_run_prompt"] and not role["seen_at_queue"]
)
counts = own_counts(
type(self).__name__,
stored_facts(text),
Role(
frozenset({"text"} if role["known"] and role["is_run_prompt"] else ())
),
Seen(text=role["seen_at_queue"]),
)
prompt_id = self._save_prompt_to_database(
text=text, category=category, tags=tags, count_run=batch_item
text=text,
category=category,
tags=tags,
count_run=bool(counts) and not self._replayed(unique_id),
)
if role["is_run_prompt"] and not prompt_id:
# The save failed: an older entry of this key must not stand in (#342)
@@ -182,6 +209,34 @@ class PromptManagerBase:
)
return prompt_id
def _cached_runs(self) -> Any:
return getattr(self, "cached_runs", CACHED_RUNS)
def _replayed(self, unique_id: Any) -> bool:
"""This job already counted the node from ComfyUI's cache (#347).
A RAM-pressure cache can free the entry after ``execution_cached``, so
the node executes in a job that replayed it: its own count is skipped.
"""
try:
return unique_id is not None and self._cached_runs().replayed(unique_id)
except Exception as error:
self.logger.debug(f"Could not check the cached-run replay: {error}")
return False
def _record_cached_item(
self, unique_id: Any, outputs: tuple, text: Any, negative_text: Any = None
) -> None:
"""Remember this execution's outputs and stored text for cached jobs (#347)."""
if unique_id is None:
return
try:
self._cached_runs().record_item(
unique_id, outputs, stored_facts(text, negative_text)
)
except Exception as error:
self.logger.debug(f"Could not record the cached-run receipt: {error}")
@staticmethod
def _run_identity(prompt_graph: Any, unique_id: Any) -> Dict[str, Any]:
"""The tracker's proof that an image belongs to this execution (#342)."""
+40 -22
View File
@@ -12,6 +12,14 @@ else:
try:
from .prompt_manager import IO, InputTypeDict, PromptManagerFocused
from .utils.cached_runs import (
PAIR,
Role,
Seen,
own_counts,
pair_components,
stored_facts,
)
from .utils.hashing import generate_prompt_hash
from .utils.pair_execution import record_pair_execution
from .utils.prompt_graph import intrinsic_pair_nodes, run_prompt_components
@@ -23,6 +31,14 @@ try:
)
except ImportError:
from prompt_manager import IO, InputTypeDict, PromptManagerFocused
from utils.cached_runs import (
PAIR,
Role,
Seen,
own_counts,
pair_components,
stored_facts,
)
from utils.hashing import generate_prompt_hash
from utils.pair_execution import record_pair_execution
from utils.prompt_graph import intrinsic_pair_nodes, run_prompt_components
@@ -165,6 +181,9 @@ class PromptManagerPair(PromptManagerFocused):
record_runtime_context(
type(self).__name__, negative_text, "", "", negative_text, role="negative"
)
self._record_cached_item(
unique_id, (positive, negative_text), text, negative_text
)
return result
def _save_and_track_pair(
@@ -209,36 +228,35 @@ class PromptManagerPair(PromptManagerFocused):
and generate_prompt_hash(resolved_negative)
== generate_prompt_hash(negative_text)
)
negative_seen = resolved_negative is not None and generate_prompt_hash(
resolved_negative
) == generate_prompt_hash(negative_text)
positive_pending = getattr(self, "pending_first_use", PENDING_FIRST_USE)
pair_pending = getattr(self, "pending_pair_first_use", PENDING_PAIR_FIRST_USE)
counted = set()
for field, component_text, resolved_text, component_id in (
("text", text, resolved_positive, prompt_id),
(
"negative_text",
negative_text,
resolved_negative,
pair["negative_prompt_id"],
),
):
component_hash = generate_prompt_hash(component_text)
if (
field not in used_fields
or not component_id
or component_hash in counted
):
facts = stored_facts(text, negative_text)
role = Role(frozenset(used_fields), pair_used)
# Own counts are skipped when this job already counted the node from
# ComfyUI's cache (#347); the pending first-use catch-up never is.
counts = (
[]
if self._replayed(unique_id)
else own_counts(
PAIR, facts, role, Seen(positive_seen, negative_seen, pair_seen)
)
)
component_ids = {"text": prompt_id, "negative_text": pair["negative_prompt_id"]}
for field, component_hash in pair_components(facts, role):
component_id = component_ids[field]
if not component_id:
continue
counted.add(component_hash)
seen_at_queue = (
resolved_text is not None
and generate_prompt_hash(resolved_text) == component_hash
runs = positive_pending.consume(component_hash) + int(
("prompt", component_hash) in counts
)
runs = positive_pending.consume(component_hash) + int(not seen_at_queue)
if runs:
self.db.record_prompt_use(component_id, times=runs)
if pair_used:
runs = pair_pending.consume(pair_run_key(text, negative_text)) + int(
not pair_seen
("pair", facts.positive_hash, facts.negative_hash) in counts
)
if runs:
self.db.record_prompt_pair_use(pair_id, times=runs)
+1
View File
@@ -239,6 +239,7 @@ class PromptManagerText(PromptManagerBase, ComfyNodeABC):
record_runtime_context(
type(self).__name__, text, prepend_text, append_text, final_text
)
self._record_cached_item(unique_id, (final_text,), storage_text)
return result
@classmethod
+53 -5
View File
@@ -5,6 +5,7 @@ import base64
import binascii
from functools import wraps
import json
import os
import re
import sqlite3
from typing import Callable
@@ -27,6 +28,7 @@ from .history_current import (
snapshot_current_output_scope,
)
from .inspection_service import CurrentFileChanged, inspect_current_file
from .images import image_category_key
try:
from ...database.execution_history import HistoryEpochChanged
@@ -39,7 +41,11 @@ try:
WindowPosition,
)
from ...utils.history_read_fence import HistoryReadFence
from ...utils.inspection_response import encode_history_inspection_response
from ...utils.inspection_response import (
OWN_CATEGORY_UNREAD,
apply_own_category,
encode_history_inspection_response,
)
except ImportError:
from database.execution_history import HistoryEpochChanged
from database.history_gallery import read_history_gallery
@@ -51,7 +57,11 @@ except ImportError:
WindowPosition,
)
from utils.history_read_fence import HistoryReadFence
from utils.inspection_response import encode_history_inspection_response
from utils.inspection_response import (
OWN_CATEGORY_UNREAD,
apply_own_category,
encode_history_inspection_response,
)
_RUN_MAX_BYTES = 8 * 1024 * 1024
@@ -677,6 +687,32 @@ class HistoryRoutesMixin:
) -> web.Response:
return await self._history_current(request, "inspection")
def _history_own_category(self, reference: dict, scope):
"""The recorded output's own category (#332), inside the snapshot.
The path is the symlink-resolved output folder plus the receipt's
subfolder and filename, keyed by ``image_category_key`` over the
configured output sources as Images keys it. A temp output or a path
no source holds has none. A failed read is OWN_CATEGORY_UNREAD.
"""
if (
reference.get("folder_type") != "output"
or scope.canonical_output is None
or candidate_locator(reference, scope) is None
):
return None
path = os.path.join(
str(scope.canonical_output), reference["subfolder"], reference["filename"]
)
try:
key = image_category_key(self._get_all_output_dirs(), path)
return None if key is None else self.db.image_category(*key)
except Exception:
self.logger.warning(
"History image category could not be read", exc_info=True
)
return OWN_CATEGORY_UNREAD
async def _history_current(
self, request: web.Request, kind: str
) -> web.StreamResponse:
@@ -701,6 +737,7 @@ class HistoryRoutesMixin:
if scope.current_scope != current_scope:
raise _current_stale()
result = db.get_history_occurrence(run_id, node, ordinal)
own = None
if result is not None:
receipt = result["receipt"]
path = candidate_locator(receipt["reference"], scope)
@@ -708,6 +745,10 @@ class HistoryRoutesMixin:
db.model.get_connection(),
(path,) if path is not None else (),
)
if kind == "inspection":
own = self._history_own_category(
receipt["reference"], scope
)
finally:
db.model.close()
if result is None:
@@ -726,9 +767,16 @@ class HistoryRoutesMixin:
resolved.canonical_target,
source_label=resolved.public["source_label"],
saved=saved.get("saved"),
library=saved.get(
"library",
{"status": "unavailable" if saved_unavailable else "none"},
library=apply_own_category(
saved.get(
"library",
{
"status": (
"unavailable" if saved_unavailable else "none"
)
},
),
own,
),
recorded=recorded_output_lookup(
db, close=True, reference=receipt["reference"]
+16 -3
View File
@@ -29,9 +29,11 @@ from .prompts import bad_request
try:
from ...database.image_categories import MAX_CATEGORY_LENGTH, MAX_KEY_LENGTH
from ...database.models import normalize_image_path
from ...utils.validators import validate_category
except ImportError:
from database.image_categories import MAX_CATEGORY_LENGTH, MAX_KEY_LENGTH
from database.models import normalize_image_path
from utils.validators import validate_category
CATEGORY_MEDIA_MESSAGE = "Only media files can have a category"
@@ -143,7 +145,7 @@ class ImageCategoryRoutesMixin:
# -- reading ---------------------------------------------------------------
def _own_category_map(self, sources, listings) -> dict:
"""``{root / listed path: own category}`` for the listed files (blocking).
"""``{normalized listed path: own category}`` for the listed files (blocking).
One primary-key range read per source; each stored row is matched to
the listing of every selected root that holds its source, so the work
@@ -162,19 +164,30 @@ class ImageCategoryRoutesMixin:
for _index, root, files in listings:
listing = self._favorite_listing(root, files)
root_key = _favorite_root_key(root)
# normalize_image_path(root / listed) without a call per file: the
# listed spellings are clean relative POSIX paths. Windows abspath
# also drops a component's trailing dots and spaces, so it keeps
# the call there.
base = os.path.join(normalize_image_path(str(root)), "")
def normalized(listed, base=base, root=root):
if os.name == "nt":
return normalize_image_path(str(root / listed))
return base + os.path.normcase(listed.replace("/", os.sep))
nested = [
item for item in _nested_sources(root, sources) if item[1] != root_key
]
for relative, category in by_source.get(root_key, {}).items():
listed = listing.get(relative)
if listed is not None and _source_prefix(relative, nested) is None:
found[root / listed] = category
found[normalized(listed)] = category
for prefix, source_key in nested:
for relative, category in by_source.get(source_key, {}).items():
full = f"{prefix}/{relative}"
listed = listing.get(full)
if listed is not None and _source_prefix(full, nested) == prefix:
found[root / listed] = category
found[normalized(listed)] = category
return found
async def _source_listings(self, roots) -> list[tuple]:
+97 -25
View File
@@ -66,6 +66,8 @@ try:
from ...utils.parallel import map_parallel
from ...utils.json_strict import dumps_strict
from ...utils.inspection_response import (
OWN_CATEGORY_UNREAD,
apply_own_category,
encode_inspection_response,
library_metadata,
)
@@ -87,6 +89,8 @@ except ImportError:
from utils.parallel import map_parallel
from utils.json_strict import dumps_strict
from utils.inspection_response import (
OWN_CATEGORY_UNREAD,
apply_own_category,
encode_inspection_response,
library_metadata,
)
@@ -565,7 +569,8 @@ class _ImageTagMatch(NamedTuple):
tags: bool # every filter tag is in its image tags or linked positive tags
query: bool # ``q`` is a case-folded substring of one image tag name
link: tuple | None # (hit, rating, positive id) of its unambiguous link
# (hit, rating, positive id, inherited category) of its unambiguous link
link: tuple | None
def _current_fingerprint(path: Path) -> tuple[int, int] | None:
@@ -1220,20 +1225,27 @@ def _admitted_entries(
links: dict | None,
favorites: frozenset | None = None,
image_tags: dict | None = None,
own_categories: dict | None = None,
):
"""Yield ``(position, entry, link match)`` for each entry the filters admit.
*links* maps normalized paths to (hit, rating, positive id) for every
unambiguously linked image (see PromptDatabase.get_image_link_matches); a
category or tag filter admits only those, and ``q`` admits a filename match
or a linked-positive hit. *favorites*, when given, admits only those listed
*links* maps normalized paths to (hit, rating, positive id, inherited
category) for every unambiguously linked image (see
PromptDatabase.get_image_link_matches); a tag filter admits only those,
and ``q`` admits a filename match or a linked-positive hit. A category
admits a file whose effective category equals it (#332 MC5): its own
category (*own_categories*, ``{normalized path: category}``) when it has one,
else the inherited one, so an unlinked file with its own category is
admitted. *favorites*, when given, admits only those listed
paths (``favorites=1``). *image_tags* maps normalized paths to their
_ImageTagMatch: a tag filter also admits a file whose current image tags
(with its linked positive's tags) carry every tag, and ``q`` also admits an
image tag name hit.
"""
query, media, folder = filters["query"], filters["media"], filters["folder"]
members_only = filters["category"] is not None or bool(filters["tags"])
category = filters["category"]
members_only = bool(filters["tags"])
own = own_categories or {}
native_folder = folder.replace("/", os.sep)
nested = native_folder + os.sep
exact = filters.get("path", "")
@@ -1252,6 +1264,12 @@ def _admitted_entries(
image = None if image_tags is None else image_tags.get(entry.normalized)
if members_only and match is None and not (image and image.tags):
continue
if category is not None:
effective = own.get(entry.normalized) if own else None
if effective is None and match is not None:
effective = match[3]
if effective != category:
continue
if (
query
and query not in entry.folded
@@ -1270,6 +1288,7 @@ def _filtered_page(
links: dict | None,
favorites: frozenset | None = None,
image_tags: dict | None = None,
own_categories: dict | None = None,
) -> tuple[int, list, int | None]:
"""Count the *entries* the filters admit and take *size* from *start* on.
@@ -1279,7 +1298,7 @@ def _filtered_page(
"""
total, page, positives = 0, [], set()
for position, entry, match in _admitted_entries(
entries, filters, links, favorites, image_tags
entries, filters, links, favorites, image_tags, own_categories
):
total += 1
if match is not None:
@@ -2219,9 +2238,10 @@ class ImageRoutesMixin:
stat'ed (never every listed file), and a candidate's tags count only
while a byte record is verified at its present ``(size, mtime_ns)``
(spec amendment S1). Under a Tags filter the candidates' linked
positives are read too, so their tags and category combine.
positives are read too, so their tags combine; the category is checked
once, on the effective category, by _admitted_entries (#332).
"""
tags, query, category = filters["tags"], filters["query"], filters["category"]
tags, query = filters["tags"], filters["query"]
wanted = set(tags)
located = self._workspace_locators(listings)
# The tag index answers which files carry a relevant name (#379); no
@@ -2251,11 +2271,9 @@ class ImageRoutesMixin:
link = facts.get(entry.normalized)
carried = set(names).union(link[4] if link else ())
matches[entry.normalized] = _ImageTagMatch(
bool(wanted)
and wanted <= carried
and (category is None or (link is not None and link[3] == category)),
bool(wanted) and wanted <= carried,
bool(query) and any(query in name.casefold() for name in names),
link[:3] if link else None,
tuple(link[:4]) if link else None,
)
return matches
@@ -2264,10 +2282,14 @@ class ImageRoutesMixin:
) -> tuple:
"""The ordered entries and what admits them (blocking; run in executor).
Returns ``(entries, key, links, image_tags, favorites)``: the listing in
the filters' order, its keyset key, and the link, image-tag and favorite
maps _admitted_entries reads. One SELECT (get_image_link_matches) reads
every linked positive when *read_links*. A failed read raises when the
Returns ``(entries, key, links, image_tags, favorites, own)``: the
listing in the filters' order, its keyset key, and the link, image-tag,
favorite and own-category maps _admitted_entries reads. One SELECT
(get_image_link_matches) reads every linked positive when *read_links*;
under a category it keeps only links whose raw recipe or positive
category is that value (the inherited one is decided in Python), and
the links of files whose own category is that value are read by path,
so their rating and positive still count (#332). A failed read raises when the
filters need it (category, tags, the rating sort) or *strict*; else the
listing stays usable without links (a search then matches filenames
only). A failed image-tag read raises under a Tags filter; a search then
@@ -2279,10 +2301,10 @@ class ImageRoutesMixin:
try:
links = self.db.get_image_link_matches(
filters["text"],
category,
tags,
every_link=True,
wanted=self._workspace_wanted(listings),
category=category,
)
except Exception:
if strict or category is not None or tags or order == "rating":
@@ -2296,6 +2318,13 @@ class ImageRoutesMixin:
),
exc_info=True,
)
own = None
if category is not None:
# A failed read raises: an unfiltered page must never pass for the
# category (DC6).
own = self._own_category_map(self._get_all_output_dirs(), listings)
if links is not None:
links = self._own_category_links(links, own, filters)
image_tags = None
if filters["query"] or tags:
try:
@@ -2342,7 +2371,30 @@ class ImageRoutesMixin:
favorites = (
self._workspace_favorite_paths(listings) if filters["favorites"] else None
)
return entries, key, links, image_tags, favorites
return entries, key, links, image_tags, favorites, own
def _own_category_links(self, links: dict, own: dict, filters: dict) -> dict:
"""*links* plus the links of files whose own category is the filter's.
Their inherited category differs, so the category's link read left
them out; read by path (get_image_link_facts), kept only when their
positive has every filter tag, as the link read would.
"""
category, wanted = filters["category"], set(filters["tags"])
missing = [
path
for path, value in own.items()
if value == category and path not in links
]
if not missing:
return links
facts = self.db.get_image_link_facts(missing, filters["text"])
extra = {
path: tuple(fact[:4])
for path, fact in facts.items()
if wanted <= set(fact[4])
}
return {**links, **extra} if extra else links
def _select_workspace_page(
self,
@@ -2371,7 +2423,7 @@ class ImageRoutesMixin:
or bool(filters["tags"])
or order == "rating"
)
entries, key, links, image_tags, favorites = self._workspace_admission(
entries, key, links, image_tags, favorites, own = self._workspace_admission(
listings, filters, read_links=needed or after is None
)
start = 0
@@ -2382,7 +2434,7 @@ class ImageRoutesMixin:
else:
start = bisect_right(entries, after, key=key)
total, page, positive_count = _filtered_page(
entries, start, limit + 1, filters, links, favorites, image_tags
entries, start, limit + 1, filters, links, favorites, image_tags, own
)
if not needed and after is not None:
positive_count = _COUNT_OMITTED
@@ -2404,12 +2456,12 @@ class ImageRoutesMixin:
The links are read strictly: the positives must be the listing's own.
"""
entries, _key, links, image_tags, favorites = self._workspace_admission(
entries, _key, links, image_tags, favorites, own = self._workspace_admission(
listings, filters, read_links=True, strict=True
)
targets, positives, unlinked, scopes = [], {}, 0, {}
for _position, entry, match in _admitted_entries(
entries, filters, links, favorites, image_tags
entries, filters, links, favorites, image_tags, own
):
if len(targets) == cap:
return ListingSelection(targets, list(positives), unlinked, True)
@@ -3728,13 +3780,20 @@ class ImageRoutesMixin:
except Exception:
# The link stays known; only its title is withheld.
self.logger.warning("Inspection link title could not be read")
# An output file's own category (#332); LoRA roots have none.
own = (
self._read_own_category(outputs, path)
if index < len(outputs)
else None
)
try:
observed = inspect_current_file(
path,
source_label=_public_path(roots[index]),
saved=saved,
library=self._inspection_library(
saved, unavailable=unavailable
library=apply_own_category(
self._inspection_library(saved, unavailable=unavailable),
own,
),
image_tags=self._inspection_image_tags(
roots[index], Path(relative).parts
@@ -3802,6 +3861,19 @@ class ImageRoutesMixin:
return read
def _read_own_category(self, sources, path):
"""One file's own category, None, or OWN_CATEGORY_UNREAD (blocking).
Keyed by ``image_category_key`` over the output sources, as Images
writes it. A failed read is logged and hides the category (DC5).
"""
try:
key = image_category_key(sources, path)
return None if key is None else self.db.image_category(*key)
except Exception:
self.logger.warning("Image category could not be read", exc_info=True)
return OWN_CATEGORY_UNREAD
def _inspection_library(self, saved: dict | None, *, unavailable: bool) -> dict:
"""The inspector's Library facts; the route sets the title afterwards."""
result = {
+165 -3
View File
@@ -11,7 +11,7 @@ import { installHostHooks } from './host-fixture.mjs';
// The worker's owned host must be up before the shared host hooks open the page.
test.beforeEach(async ({ execution }) => { expect(execution.databasePath).toBeTruthy(); });
installHostHooks(test, ['PromptManager', 'PromptManagerText', 'PromptManagerFocused', 'PromptManagerPair', 'PMHistoryInputs']);
installHostHooks(test, ['PromptManager', 'PromptManagerText', 'PromptManagerFocused', 'PromptManagerPair', 'PMHistoryInputs', 'PMCycledText']);
// `primitive`: a core string node the queue-time resolver follows. `legacy`: the frontend's virtual Primitive node,
// whose value is queued as literal text. `preview`: PrimitiveString -> PreviewAny, the shape of the maintainer's
@@ -37,6 +37,7 @@ function build(page, { kind, positive, negative, prefix }) {
primitive.widgets[0].value = side.text;
return;
}
if (side.via === 'cycled') { connect(add('PMCycledText', 0, y), 0, prompt, name); return; }
const text = add('PrimitiveString', 0, y);
text.widgets.find(widget => widget.name === 'value').value = side.text;
let source = text;
@@ -75,9 +76,9 @@ function build(page, { kind, positive, negative, prefix }) {
}, { kind, positive, negative, prefix });
}
async function run(execution, page, testCase) {
async function run(execution, page, testCase, built) {
const prefix = `textlink/${randomUUID()}`;
const { output, workflow, promptNode } = await build(page, { ...testCase, prefix });
const { output, workflow, promptNode } = built ?? await build(page, { ...testCase, prefix });
const id = randomUUID();
const response = await execution.request.post('/prompt', { data: { prompt: output, prompt_id: id, extra_data: { extra_pnginfo: { workflow } } } });
expect(response.ok(), await response.text()).toBe(true);
@@ -193,3 +194,164 @@ for (const [label, linked, negativeVia] of [['positive', ['text'], 'primitive'],
await expectImagesLinked(execution, evidence, sides.text.via === 'preview');
});
}
// #347: a cached PromptManager node counts every job, as an executing one does. Jobs after the first change only
// the sampler seed, so ComfyUI serves the PromptManager node (and its PreviewAny source) from its output cache.
const COUNTS = `import json, sqlite3, sys, pathlib
db = sqlite3.connect(pathlib.Path(sys.argv[1]).resolve().as_uri() + "?mode=ro", uri=True)
one = lambda sql, key: (db.execute(sql, (int(key),)).fetchone() or [None])[0] if key != "-" else None
print(json.dumps({"positive": one("SELECT run_count FROM prompts WHERE id=?", sys.argv[2]),
"negative": one("SELECT run_count FROM prompts WHERE id=?", sys.argv[3]),
"pair": one("SELECT run_count FROM prompt_pairs WHERE id=?", sys.argv[4])}))`;
async function runCounts(execution, context) {
const key = value => value == null ? '-' : String(value);
const { stdout } = await promisify(execFile)(process.env.PM_TEST_PYTHON || 'python',
['-c', COUNTS, execution.databasePath, key(context.positive_prompt_id), key(context.negative_prompt_id), key(context.pair_id)], { timeout: 5000 });
return JSON.parse(stdout);
}
async function setCache(execution, mode) {
expect((await execution.request.post('/pm_execution_fixture/cache', { data: { mode } })).ok()).toBe(true);
}
// Queue an already built workflow again with another sampler seed; returns the job's history item and cached nodes.
async function queueAgain(execution, first, { seed, edit, targets, status = 'success' } = {}) {
const output = structuredClone(first.output);
const sampler = Object.values(output).find(node => node.class_type === 'KSampler');
sampler.inputs.seed = seed;
edit?.(output);
const id = randomUUID();
const data = { prompt: output, prompt_id: id, extra_data: { extra_pnginfo: { workflow: first.workflow } } };
if (targets) data.partial_execution_targets = targets;
const response = await execution.request.post('/prompt', { data });
expect(response.ok(), await response.text()).toBe(true);
let item;
await expect.poll(async () => {
item = (await (await execution.request.get(`/history/${id}`)).json())[id];
return item?.status?.status_str ?? null;
}, { timeout: 20000 }).toBe(status);
const cached = item.status.messages.find(([event]) => event === 'execution_cached')?.[1]?.nodes ?? [];
return { id, item, cached };
}
// The recorded context of one job, once its capture reached SQLite.
async function capturedContext(execution, id) {
let result;
await expect.poll(async () => {
result = await readExecutionRuns(execution.databasePath, id);
const row = result.rows[0];
return row ? result.captures[row.run_id].state : null;
}, { timeout: 10000 }).toMatch(/^(complete|partial)$/);
return result.captures[result.rows[0].run_id].contexts[0];
}
async function firstRun(execution, page, testCase) {
const prefix = `textlink/${randomUUID()}`;
const built = await build(page, { ...testCase, prefix });
const evidence = await run(execution, page, testCase, built);
return { ...evidence, workflow: built.workflow };
}
for (const kind of ['PromptManager', 'PromptManagerFocused', 'PromptManagerSplit', 'PromptManagerText', 'PromptManagerPair']) {
test(`${kind} counts each cached re-run of text linked from PreviewAny (#347)`, async ({ execution, page }) => {
test.setTimeout(90000);
await setCache(execution, 'RAM_PRESSURE');
const pair = kind === 'PromptManagerPair';
const testCase = { kind, positive: { typed: 'Typed under the link', text: unique(`Cached ${kind}`), via: 'preview' },
negative: pair ? { typed: unique('Typed negative') } : undefined };
const first = await firstRun(execution, page, testCase);
for (const seed of [2, 3]) {
const again = await queueAgain(execution, first, { seed });
expect(again.cached, 'ComfyUI served the PromptManager node from its cache').toContain(first.promptNode);
}
const counts = await runCounts(execution, first.context);
expect(counts.positive).toBe(3);
if (pair) expect(counts).toMatchObject({ negative: 0, pair: 3 });
});
}
test.describe('cached runs in one renderer (#347)', () => {
// Cache behavior does not depend on the renderer: one project runs these.
test.beforeEach(({}, testInfo) => { test.skip(testInfo.project.name !== 'classic-start', 'covered in classic-start'); });
const linked = () => ({ kind: 'PromptManager', positive: { typed: 'Typed under the link', text: unique('Cached'), via: 'preview' } });
for (const mode of ['CLASSIC', 'LRU', 'RAM_PRESSURE', 'NONE']) {
test(`counts 3 jobs under the ${mode} cache`, async ({ execution, page }) => {
test.setTimeout(90000);
await setCache(execution, mode);
try {
const first = await firstRun(execution, page, linked());
for (const seed of [2, 3]) {
const again = await queueAgain(execution, first, { seed });
if (mode === 'NONE') expect(again.cached).not.toContain(first.promptNode);
else expect(again.cached).toContain(first.promptNode);
}
expect((await runCounts(execution, first.context)).positive).toBe(3);
} finally { await setCache(execution, 'RAM_PRESSURE'); }
});
}
test('alternating upstream values A, B, A under LRU each count their own prompt', async ({ execution, page }) => {
test.setTimeout(90000);
await setCache(execution, 'LRU');
try {
const [a, b] = [unique('Cycled A'), unique('Cycled B')];
const cycle = async value => expect((await execution.request.post('/pm_execution_fixture/cycled', { data: { value } })).ok()).toBe(true);
await cycle(a);
const first = await firstRun(execution, page, { kind: 'PromptManager', positive: { typed: 'Typed under the link', text: a, via: 'cycled' } });
expect(first.context.positive.raw).toBe(a);
await cycle(b);
const second = await queueAgain(execution, first, { seed: 2 });
expect(second.cached).not.toContain(first.promptNode);
await cycle(a);
const third = await queueAgain(execution, first, { seed: 3 });
expect(third.cached, 'LRU kept A').toContain(first.promptNode);
expect((await runCounts(execution, first.context)).positive).toBe(2);
const bContext = await capturedContext(execution, second.id);
expect(bContext.positive.raw).toBe(b);
expect((await runCounts(execution, bContext)).positive).toBe(1);
} finally { await setCache(execution, 'RAM_PRESSURE'); }
});
test('typed text served from the cache still counts once per job', async ({ execution, page }) => {
test.setTimeout(90000);
await setCache(execution, 'RAM_PRESSURE');
const first = await firstRun(execution, page, { kind: 'PromptManager', positive: { typed: unique('Typed cached') } });
for (const seed of [2, 3]) expect((await queueAgain(execution, first, { seed })).cached).toContain(first.promptNode);
expect((await runCounts(execution, first.context)).positive).toBe(3);
});
test('a job failing downstream of a cached node counts it once; a job that does not need it, never', async ({ execution, page }) => {
test.setTimeout(90000);
await setCache(execution, 'RAM_PRESSURE');
const first = await firstRun(execution, page, linked());
const failing = await queueAgain(execution, first, { seed: 2, status: 'error', edit: output => {
output.pm_failure = { class_type: 'PMHistoryOutput', inputs: { value: [Object.keys(output).find(id => output[id].class_type === 'KSampler'), 0], mode: 'failure' } };
} });
expect(failing.cached).toContain(first.promptNode);
expect((await runCounts(execution, first.context)).positive).toBe(2);
// "Run selected output": only an unrelated output node; ComfyUI still reports the cached PromptManager node
const unrelated = await queueAgain(execution, first, { seed: 2, targets: ['pm_unrelated'], edit: output => {
output.pm_unrelated = { class_type: 'PMHistoryCached', inputs: { value: unique('unrelated') } };
} });
expect(unrelated.cached).toContain(first.promptNode);
expect((await runCounts(execution, first.context)).positive).toBe(2);
});
test('a cached node rewired to the negative input does not count the positive prompt', async ({ execution, page }) => {
test.setTimeout(90000);
await setCache(execution, 'RAM_PRESSURE');
const first = await firstRun(execution, page, linked());
const rewired = await queueAgain(execution, first, { seed: 2, edit: output => {
const sampler = Object.values(output).find(node => node.class_type === 'KSampler');
[sampler.inputs.positive, sampler.inputs.negative] = [sampler.inputs.negative, sampler.inputs.positive];
} });
expect(rewired.cached).toContain(first.promptNode);
expect((await runCounts(execution, first.context)).positive).toBe(1);
// The history capture of a cached job is unchanged: effective text from the cache, no stored-prompt hint
const cachedContext = await capturedContext(execution, (await queueAgain(execution, first, { seed: 3 })).id);
expect(cachedContext.positive.effective).toBe(first.context.positive.effective);
expect(cachedContext.positive_prompt_id).toBeNull();
expect((await runCounts(execution, first.context)).positive).toBe(2);
});
});
+33
View File
@@ -201,6 +201,39 @@ NODE_CLASS_MAPPINGS.update(
{"PMHistoryImages": PMHistoryImages, "PMHistoryImageOutput": PMHistoryImageOutput}
)
# #347: text whose value (and cache signature) follows a value the test sets, so
# one queued graph can serve A, then B, then A again (each cached under LRU).
_CYCLED = {"value": "cycled"}
class PMCycledText:
@classmethod
def INPUT_TYPES(cls) -> dict:
return {"required": {}}
RETURN_TYPES = ("STRING",)
FUNCTION = "execute"
CATEGORY = "testing/execution-history"
@classmethod
def IS_CHANGED(cls, **kwargs: object) -> str:
return _CYCLED["value"]
def execute(self) -> tuple:
return (_CYCLED["value"],)
NODE_CLASS_MAPPINGS["PMCycledText"] = PMCycledText
@PromptServer.instance.routes.post("/pm_execution_fixture/cycled")
async def cycled_text(request: web.Request) -> web.Response:
value = (await request.json())["value"]
if type(value) is not str:
raise web.HTTPBadRequest()
_CYCLED["value"] = value
return web.json_response({"value": value})
@PromptServer.instance.routes.post("/pm_execution_fixture/metadata")
async def metadata(request: web.Request) -> web.Response:
+29
View File
@@ -383,3 +383,32 @@ test('the D117 wrap rule targets only the settings summary button', async () =>
assert.deepEqual(rules.map(line => line.trim()), ['.ca-tools #viewer-details .settings-summary { white-space: normal; /* D117 */ }']);
assert.doesNotMatch(css, /#viewer-details > div > button/);
});
test('category chip names its origin and shows an own category for any link status (#332)', async () => {
const base = await localRecord(localPng('a.png'));
const chipTexts = record => {
const {f, done} = renderFull(record);
try {
return f.walk(f.container, n => /(^| )chip( |$)/.test(n.className ?? '')).map(n => n.textContent);
} finally { done(); }
};
const cases = [
[{status:'none', category:'Scans', category_scope:'image'}, 'Image category: Scans'],
[{status:'ambiguous', category:'Scans', category_scope:'image'}, 'Image category: Scans'],
[{status:'linked', category:'Portraits', category_scope:'image'}, 'Image category: Portraits'],
[{status:'linked', category:'Landscapes', category_scope:'positive'}, 'Positive prompt category: Landscapes'],
[{status:'linked', category:'Recipes', category_scope:'pair'}, 'Recipe category: Recipes'],
];
for (const [library, text] of cases) {
const chips = chipTexts({...base, library:{...base.library, tags:[], tags_scope:null, ...library}});
assert.ok(chips.includes(text), `${JSON.stringify(library)} -> ${chips}`);
}
for (const library of [
{status:'none', category:'Inherited', category_scope:'positive'},
{status:'linked', category:null, category_scope:null},
{status:'none', category:null, category_scope:null},
]) {
const chips = chipTexts({...base, library:{...base.library, tags:[], tags_scope:null, ...library}});
assert.equal(chips.some(chip => /category: /.test(chip)), false, JSON.stringify(library));
}
});
@@ -278,6 +278,7 @@ test('record rejects root path media and invalid versioned fields with retryable
data => { data.resolution.sampler_id = {}; },
data => { data.library.tags_scope = 'recipe'; },
data => { data.library.category_scope = 'recipe'; },
data => { data.library.tags_scope = 'image'; },
data => { delete data.library.category_scope; },
data => { delete data.library.tags_scope; },
]) {
@@ -942,3 +943,14 @@ test("names the server quotes (parentheses, '!*') still match their original and
assert.equal(record.download.url, entry.url);
assert.deepEqual(record.preview, { url: entry.thumbnail_url });
});
test('record accepts an image\'s own category on any link status (#332)', () => {
const toRecord = contract('toInspectionRecord');
for (const status of ['linked', 'none', 'ambiguous', 'unavailable']) {
const data = envelope();
Object.assign(data.library, { status, category: 'Scans', category_scope: 'image' });
const record = toRecord(data, image(), extractGenerationFacts);
assert.equal(record.library.category, 'Scans');
assert.equal(record.library.category_scope, 'image');
}
});
+74 -2
View File
@@ -62,12 +62,16 @@ class TestVocabularyAPI(ImageAPITestCase):
},
)
self.assertEqual(
categories["categories"], [{"value": "Animal", "prompt_count": 1}]
categories["categories"],
[{"value": "Animal", "prompt_count": 1, "image_count": 0}],
)
_, rest = await self.get(
"categories", limit=1, cursor=categories["next_cursor"]
)
self.assertEqual(rest["categories"], [{"value": "Pet", "prompt_count": 1}])
self.assertEqual(
rest["categories"],
[{"value": "Pet", "prompt_count": 1, "image_count": 0}],
)
self.assertFalse(rest["has_more"])
async def test_vocabulary_cursor_is_signed_epoch_scoped(self):
@@ -488,6 +492,7 @@ class TestVocabularyAPI(ImageAPITestCase):
"other_prompts": 0,
"negative_variants": 1,
"archived_negative_variants": 0,
"images": 0,
}
self.assertEqual(
preview,
@@ -528,6 +533,73 @@ class TestVocabularyAPI(ImageAPITestCase):
self.assertEqual(response.status, 409)
self.assertEqual(replay["code"], "vocabulary-conflict")
async def test_category_rows_report_images_and_image_only_values(self):
self._save_prompt("Owner", category="Prompt value")
_, options_before = await self.get_options()
self.api.db.set_image_categories(
[("root", "a.png"), ("root", "b.png")], "Prompt value"
)
self.api.db.set_image_categories([("root", "c.png")], "Image value")
response, first = await self.get("categories", limit=1)
self.assertEqual(response.status, 200)
self.assertEqual(
first["categories"],
[{"value": "Prompt value", "prompt_count": 1, "image_count": 2}],
)
self.assertEqual(first["total"], 2)
_, rest = await self.get("categories", limit=1, cursor=first["next_cursor"])
self.assertEqual(
rest["categories"],
[{"value": "Image value", "prompt_count": 0, "image_count": 1}],
)
self.assertFalse(rest["has_more"])
# The Prompt library's category list does not gain image values.
_, options_after = await self.get_options()
self.assertEqual(options_after, options_before)
self.assertEqual(options_after["categories"], ["Prompt value"])
async def get_options(self):
response = await self.client.get("/prompt_manager/library/options")
return response, await response.json()
async def test_category_routes_rename_image_only_value(self):
epoch = self.api.db.get_execution_epoch()
self._save_prompt("Bystander", category="Kept")
self.api.db.set_image_categories(
[("root", "a.png"), ("second", "b.png")], "Image value"
)
prompts_before = self.category_rows()
body = {"expected_epoch": epoch, "source": "Image value", "target": "Moved"}
response, preview = await self.send("POST", "categories/preview", body)
self.assertEqual(response.status, 200, preview)
impact = {
"positive_prompts": 0,
"other_prompts": 0,
"negative_variants": 0,
"archived_negative_variants": 0,
"images": 2,
}
self.assertEqual(preview["impact"], impact)
stale = preview["impact_token"]
self.api.db.set_image_categories([("root", "late.png")], "Image value")
response, refused = await self.send(
"POST", "categories/apply", {**body, "impact_token": stale}
)
self.assertEqual(response.status, 409)
self.assertEqual(refused["code"], "vocabulary-conflict")
self.assertEqual(self.api.db.image_category("root", "a.png"), "Image value")
_, preview = await self.send("POST", "categories/preview", body)
response, applied = await self.send(
"POST",
"categories/apply",
{**body, "impact_token": preview["impact_token"]},
)
self.assertEqual(response.status, 200, applied)
self.assertEqual(applied["impact"], {**impact, "images": 3})
for key in (("root", "a.png"), ("second", "b.png"), ("root", "late.png")):
self.assertEqual(self.api.db.image_category(*key), "Moved")
self.assertEqual(self.category_rows(), prompts_before)
async def test_preview_validation(self):
epoch = self.api.db.get_execution_epoch()
self._save_prompt("Owner", category="Scenery")
@@ -0,0 +1,262 @@
"""Images filters by the effective category (#332 MC4, MC5, DC6).
An image's effective category is its own category when it has one, else the
category its link supplies (the recipe's when set, else the positive's), else
none. The category filter, "select all N matching" and saved views match it.
"""
import os
import sqlite3
from unittest.mock import patch
try:
import test_api_workspace_images as workspace
from test_image_categories import category_checksum
except ImportError:
from tests import test_api_workspace_images as workspace
from tests.test_image_categories import category_checksum
from py.api.images import image_category_key, parse_listing_filter
class WorkspaceCategoryCase(workspace.ImageAPITestCase):
"""Listing helpers plus own-category writes keyed as the routes key them."""
media = workspace.TestWorkspaceImages.media
get = workspace.TestWorkspaceImages.get
tag = workspace.TestWorkspaceImageTags.tag
names = workspace.TestWorkspaceImageTags.names
async def tearDownAsync(self):
self.api.db.close_all()
workspace.assert_closed(self, self._temp_db.name, self._output_tmp.name)
await super().tearDownAsync()
def own(self, path, value, sources=None):
sources = sources or self.api._get_all_output_dirs()
self.api.db.set_image_categories([image_category_key(sources, path)], value)
async def listed(self, **params):
response, body = await self.get(sort="name", **params)
self.assertEqual(response.status, 200, body)
return self.names(body), body
class TestEffectiveCategoryFilter(WorkspaceCategoryCase):
def batch(self):
"""Four images of "castle at dusk" (Landscapes); one is overridden."""
castle = self._save_prompt("castle at dusk", category="Landscapes", rating=4)
files = [self.media(f"castle_{n}.png", mtime=100 + n) for n in range(4)]
for path in files:
self.api.db.link_image_to_prompt(castle, str(path))
self.own(files[0], "Portraits")
return castle, files
async def test_category_filter_prefers_own_category(self):
castle, _files = self.batch()
conn = self.api.db.model.get_connection()
before = category_checksum(conn)
names, body = await self.listed(category="Portraits")
self.assertEqual(names, ["castle_0.png"])
# The caption still counts the prompt behind the image shown.
self.assertEqual(body["positive_count"], 1)
names, body = await self.listed(category="Landscapes")
self.assertEqual(names, ["castle_1.png", "castle_2.png", "castle_3.png"])
self.assertEqual(body["total"], 3)
names, _ = await self.listed()
self.assertEqual(len(names), 4)
self.assertEqual(category_checksum(conn), before)
self.assertEqual(self.api.db.get_prompt_by_id(castle)["category"], "Landscapes")
async def test_category_filter_lists_unlinked_with_own(self):
scans = [self.media(f"scan_{n}.png", mtime=100 + n) for n in range(2)]
self.media("plain.png", mtime=50)
for path in scans:
self.own(path, "Scans")
names, body = await self.listed(category="Scans")
self.assertEqual(names, ["scan_0.png", "scan_1.png"])
self.assertEqual(body["positive_count"], 0)
async def test_recipe_category_is_inherited(self):
"""MC4: a recipe's own category, not the positive's, is inherited."""
db = self.api.db
self._save_prompt("Pair positive", category="Night")
pair = db.save_prompt_pair("Pair positive", "Noise", category="Recipes")
blank = db.save_prompt_pair("Pair positive", "Other noise", category=" ")
recipe = self.media("recipe.png", mtime=103)
blank_recipe = self.media("blank_recipe.png", mtime=102)
positive = db.get_prompt_pair(pair)["positive_prompt_id"]
for path, pair_id in ((recipe, pair), (blank_recipe, blank)):
image = db.link_image_to_prompt(positive, str(path))
db.link_image_to_pair(pair_id, image, {"source": "fixture"})
self.assertEqual((await self.listed(category="Recipes"))[0], ["recipe.png"])
# A blank recipe category falls back to the positive's.
self.assertEqual((await self.listed(category="Night"))[0], ["blank_recipe.png"])
# An own category still wins over the recipe's.
self.own(recipe, "Mine")
self.assertEqual((await self.listed(category="Recipes"))[0], [])
self.assertEqual((await self.listed(category="Mine"))[0], ["recipe.png"])
async def test_ambiguous_files_inherit_nothing_but_keep_their_own(self):
first = self._save_prompt("Owner one", category="Night")
second = self._save_prompt("Owner two", category="Night")
ambiguous = self.media("ambiguous.png")
for owner in (first, second):
self.api.db.link_image_to_prompt(owner, str(ambiguous))
self.assertEqual((await self.listed(category="Night"))[0], [])
self.own(ambiguous, "Night")
self.assertEqual((await self.listed(category="Night"))[0], ["ambiguous.png"])
async def test_combines_with_tags_search_and_rating(self):
stone = self._save_prompt(
"castle at dusk", category="Landscapes", tags=["stone"], rating=4
)
files = [self.media(f"castle_{n}.png", mtime=100 + n) for n in range(2)]
for path in files:
self.api.db.link_image_to_prompt(stone, str(path))
self.own(files[0], "Portraits")
other = self._save_prompt("harbor at noon", category="Portraits", rating=2)
harbor = self.media("harbor.png", mtime=90)
self.api.db.link_image_to_prompt(other, str(harbor))
names, _ = await self.listed(category="Portraits", tags="stone")
self.assertEqual(names, ["castle_0.png"])
names, _ = await self.listed(category="Portraits", q="harbor")
self.assertEqual(names, ["harbor.png"])
names, _ = await self.listed(category="Portraits", q="castle_0")
self.assertEqual(names, ["castle_0.png"])
response, body = await self.get(category="Portraits", sort="rating")
self.assertEqual(response.status, 200, body)
self.assertEqual(self.names(body), ["castle_0.png", "harbor.png"])
self.assertEqual(body["positive_count"], 2)
async def test_image_tag_admitted_file_keeps_its_inherited_category(self):
night = self._save_prompt("Night owl", category="Night", tags=["a"])
tagged = self.media("tagged.png", mtime=103)
unlinked = self.media("unlinked.png", mtime=102)
self.api.db.link_image_to_prompt(night, str(tagged))
self.tag(tagged, ["b"])
self.tag(unlinked, ["a", "b"])
names, _ = await self.listed(tags="a,b", category="Night")
self.assertEqual(names, ["tagged.png"])
self.own(unlinked, "Night")
names, _ = await self.listed(tags="a,b", category="Night")
self.assertEqual(names, ["tagged.png", "unlinked.png"])
async def test_cursor_pages_walk_only_the_category(self):
files = [self.media(f"img_{n:02}.png", mtime=100 + n) for n in range(7)]
for path in files[::2]:
self.own(path, "Even")
seen, cursor = [], None
while True:
params = {"category": "Even", "limit": 2, "sort": "name"}
if cursor:
params["cursor"] = cursor
response, body = await self.get(**params)
self.assertEqual(response.status, 200, body)
seen += self.names(body)
self.assertEqual(body["total"], 4)
cursor = body["next_cursor"]
if not cursor:
break
self.assertEqual(seen, ["img_00.png", "img_02.png", "img_04.png", "img_06.png"])
async def test_an_outer_source_lists_a_nested_sources_own_category(self):
inner = self.output_dir / "nested"
nested = self.media("nested/a.png")
self.media("b.png")
self.api._get_all_output_dirs = lambda: [self.output_dir, inner]
self.own(nested, "Scans")
names, _ = await self.listed(category="Scans", root="0")
self.assertEqual(names, ["nested/a.png"])
names, _ = await self.listed(category="Scans", root="1")
self.assertEqual(names, ["a.png"])
# The outer source's own key for that path is not the file's key.
from py.api.images import _favorite_root_key
self.api.db.clear_image_categories([image_category_key([inner], nested)])
self.api.db.set_image_categories(
[(_favorite_root_key(self.output_dir), "nested/a.png")], "Stale"
)
self.assertEqual((await self.listed(category="Stale"))[0], [])
async def test_a_gone_files_category_is_not_listed(self):
gone = self.media("gone.png")
self.own(gone, "Scans")
os.unlink(gone)
self.api.invalidate_gallery_cache()
self.assertEqual((await self.listed(category="Scans"))[0], [])
async def test_a_failed_own_category_read_is_listing_failed(self):
self.media("a.png")
def broken(*_args, **_kwargs):
raise sqlite3.OperationalError("secret detail")
with patch.object(self.api.db, "own_categories_of", side_effect=broken):
with self.assertLogs("prompt_manager", level="ERROR"):
response, body = await self.get(category="Scans")
self.assertEqual(response.status, 500)
self.assertNotIn("secret", str(body))
# Without a category nothing new is read.
response, body = await self.get()
self.assertEqual(response.status, 200, body)
async def test_select_all_matching_uses_the_effective_category(self):
_castle, files = self.batch()
resolved = await self.api.resolve_listing_filter(
parse_listing_filter({"category": "Portraits"}), 100
)
self.assertEqual([t[2] for t in resolved.targets], ["castle_0.png"])
resolved = await self.api.resolve_listing_filter(
parse_listing_filter({"category": "Landscapes"}), 100
)
self.assertEqual(len(resolved.targets), 3)
self.assertEqual(len(files), 4)
async def test_saved_view_category_means_the_effective_category(self):
"""CUJ 8: a saved view's category is the listing's query value."""
_castle, files = self.batch()
self.own(files[1], "Portraits")
names, _ = await self.listed(category="Portraits")
self.assertEqual(names, ["castle_0.png", "castle_1.png"])
class TestFilterAndChipAgreeOnBlanks(WorkspaceCategoryCase):
"""The filter and the inspector use one inherited-category rule."""
async def chip(self, path):
response = await self.client.get(
f"/prompt_manager/images/inspect/{path}", params={"root": "0"}
)
body = await response.json()
self.assertEqual(response.status, 200, body)
return body["library"]["category"]
async def test_blank_recipe_values_agree(self):
db = self.api.db
positive = self._save_prompt("Pair positive", category="Night")
blanks = {"spaces": " ", "tab": "\t", "nbsp": " ", "set": "Recipes"}
for index, (label, value) in enumerate(blanks.items()):
pair = db.save_prompt_pair("Pair positive", f"noise {label}")
path = self.media(f"{label}.png", mtime=100 + index)
image = db.link_image_to_prompt(positive, str(path))
db.link_image_to_pair(pair, image, {"source": "fixture"})
with db.model.get_connection() as conn:
conn.execute(
"UPDATE prompt_pairs SET category=? WHERE id=?", (value, pair)
)
blob = db.save_prompt_pair("Pair positive", "noise blob")
path = self.media("blob.png", mtime=90)
image = db.link_image_to_prompt(positive, str(path))
db.link_image_to_pair(blob, image, {"source": "fixture"})
with db.model.get_connection() as conn:
conn.execute(
"UPDATE prompt_pairs SET category=CAST(' ' AS BLOB) WHERE id=?",
(blob,),
)
for name in ("spaces", "tab", "nbsp", "set", "blob"):
with self.subTest(name=name):
shown = await self.chip(f"{name}.png")
for value in ("Night", "Recipes"):
listed = f"{name}.png" in (await self.listed(category=value))[0]
self.assertEqual(listed, shown == value, (name, value, shown))
+69 -13
View File
@@ -1347,7 +1347,7 @@ class TestWorkspaceImages(ImageAPITestCase):
_, data = await self.get(q="harbor")
self.assertEqual(self._names(data), ["night_a.png"])
self.assertEqual(
self.api.db.get_image_paths_matching("zebra", None, []),
self.api.db.get_image_paths_matching("zebra", []),
{
normalize_image_path(str(files[name]))
for name in ("night_ab", "day_ab", "titled")
@@ -1407,8 +1407,10 @@ class TestWorkspaceImages(ImageAPITestCase):
for query in statements
if query.lstrip().upper().startswith(("SELECT", "WITH"))
]
# Plus the one stream of tagged image records q and tags read (B3).
self.assertEqual(len(selects), 4, selects)
# Plus the one stream of tagged image records q and tags read (B3),
# and the category's own-category read of the one source (#332).
self.assertEqual(len(selects), 5, selects)
self.assertEqual(sum("image_categories" in q for q in selects), 1, selects)
self.assertEqual(sum("image_favorites" in q for q in selects), 1, selects)
self.assertEqual(sum("media_image_tags" in q for q in selects), 1, selects)
@@ -1528,15 +1530,16 @@ class TestWorkspaceImages(ImageAPITestCase):
elsewhere = self.api.db.get_prompt_id_for_image("/elsewhere/7.png")
self.assertEqual(
self.api.db.get_image_link_matches(
"", "Night", [], every_link=True, wanted=wanted
"", [], every_link=True, wanted=wanted, category="Night"
),
{
normalize_image_path(str(files["night_ab"])): (
True,
5,
self.api.db.get_prompt_id_for_image(str(files["night_ab"])),
"Night",
),
"/elsewhere/7.png": (True, 5, elsewhere),
"/elsewhere/7.png": (True, 5, elsewhere, "Night"),
},
)
@@ -1588,6 +1591,24 @@ class TestWorkspaceImages(ImageAPITestCase):
for i in range(links)
],
)
# 2,000 images (one in 25 at 50,000 files) carry their own category
# (#332), so a category filter also reads and matches those.
from py.api.images import _favorite_root_key
root_key = _favorite_root_key(self.output_dir)
with self.api.db.model.get_connection() as conn:
conn.executemany(
"INSERT INTO image_categories VALUES(?,?,?,?)",
[
(
root_key,
listed[i][0].relative_to(self.output_dir).as_posix(),
f"cat-{i % 10}",
"2026-10-27T00:00:00+00:00",
)
for i in range(1, files, 25)[:2_000]
],
)
return listed
_BUDGET_REQUESTS = (
@@ -1615,7 +1636,7 @@ class TestWorkspaceImages(ImageAPITestCase):
import math
from py.api import images as images_module
from database.operations import link_filter_query
from database.operations import link_filter_query, link_matches_json_query
listed = await asyncio.get_running_loop().run_in_executor(
None, self._budget_fixture, 5_000, 2_000
@@ -1678,7 +1699,13 @@ class TestWorkspaceImages(ImageAPITestCase):
# q and tags also stream the tagged image records once
# (#249 B3); none are tagged here, so nothing is stat'ed.
streamed = "q" in params or "tags" in params
self.assertEqual(len(found), 2 if streamed else 1, found)
# A category also reads the own categories, one primary
# key range per source (#332).
own = "category" in params
self.assertEqual(len(found), 1 + streamed + own, found)
self.assertEqual(
sum("image_categories" in query for query in found), int(own)
)
self.assertEqual(
sum("media_image_tags" in query for query in found),
int(streamed),
@@ -1738,13 +1765,42 @@ class TestWorkspaceImages(ImageAPITestCase):
self.assertEqual(self.api._workspace_listed_cache, {})
self.assertEqual(self.api._workspace_order_cache, {})
# The per-row ambiguity probe is an index search, never a scan.
sql, params = link_filter_query("lantern", "cat-3", ["tag-1"], every_link=True)
with self.api.db.model.get_connection() as conn:
plan = [row[3] for row in conn.execute("EXPLAIN QUERY PLAN " + sql, params)]
self.assertIn(
"SEARCH other USING COVERING INDEX idx_image_file_path (file_path=?)", plan
# The read runs it wrapped in one JSON aggregate (#354): same plan.
sql, params = link_filter_query("lantern", ["tag-1"], every_link=True)
for query in (sql, link_matches_json_query(sql)):
with self.subTest(wrapped=query is not sql):
with self.api.db.model.get_connection() as conn:
plan = [
row[3]
for row in conn.execute("EXPLAIN QUERY PLAN " + query, params)
]
self.assertIn(
"SEARCH other USING COVERING INDEX idx_image_file_path "
"(file_path=?)",
plan,
)
self.assertFalse(
[step for step in plan if step.startswith("SCAN other")]
)
# The category's narrowing clause (#332) keeps every join an index
# search, plain and wrapped with the raw categories.
sql, params = link_filter_query(
"", ["tag-1"], every_link=True, category="cat-3"
)
self.assertFalse([step for step in plan if step.startswith("SCAN other")])
for query in (sql, link_matches_json_query(sql, categories=True)):
with self.subTest(category_wrapped=query is not sql):
with self.api.db.model.get_connection() as conn:
plan = [
row[3]
for row in conn.execute("EXPLAIN QUERY PLAN " + query, params)
]
self.assertIn(
"SEARCH other USING COVERING INDEX idx_image_file_path "
"(file_path=?)",
plan,
)
scans = [step for step in plan if step.startswith("SCAN")]
self.assertTrue(all(step.startswith("SCAN gi") for step in scans), plan)
@unittest.skipUnless(
PERF_BUDGET and not TRACED,
+823
View File
@@ -0,0 +1,823 @@
"""Cached re-runs of runtime-linked prompt text count once per job (#347).
Jobs are simulated end to end: the queue hook (submit + dequeue), the
``execution_cached`` replay, the real node executions and ComfyUI's output
cache (fakes of ``execution`` and ``comfy_execution.caching``, as in
tests/test_execution_context_host.py).
"""
import logging
import os
import sys
import tempfile
import threading
import types
import unittest
from collections import namedtuple
from unittest import mock
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
try:
from tests.open_handles import assert_closed
except ImportError: # discovered with tests/ as the top-level directory
from open_handles import assert_closed
from database.operations import PromptDatabase
from utils import cached_runs as module
from utils.cached_runs import (
CachedRuns,
Facts,
Role,
Seen,
graph_role,
graph_seen,
own_counts,
register_cached_runs,
stored_facts,
)
from utils.execution_context_host import cached_string_outputs
from utils.hashing import generate_prompt_hash
from utils.run_keys import prompt_node_key, upstream_closure, upstream_nodes
from utils.usage_tracking import (
PendingFirstUse,
QueuedRuns,
find_pair_id,
find_pair_id_by_hashes,
handle_queued_prompt,
on_dequeue,
)
Entry = namedtuple("CacheEntry", "ui outputs")
LLM = "a lighthouse the LLM wrote"
class _Cache:
def __init__(self):
self.entries = {}
def get_local(self, node_id):
return self.entries.get(node_id)
class HierarchicalCache(_Cache):
pass
class LRUCache(_Cache):
pass
class RAMPressureCache(LRUCache):
pass
class NullCache(_Cache):
pass
def host_modules():
"""Fake ``execution`` and ``comfy_execution.caching`` for the cache reader."""
execution = types.ModuleType("execution")
execution.CacheEntry = Entry
class PromptExecutor:
def add_message(self, event, data, broadcast):
return None
execution.PromptExecutor = PromptExecutor
caching = types.ModuleType("comfy_execution.caching")
for cls in (HierarchicalCache, LRUCache, RAMPressureCache, NullCache):
setattr(caching, cls.__name__, cls)
package = types.ModuleType("comfy_execution")
package.caching = caching
return {
"execution": execution,
"comfy_execution": package,
"comfy_execution.caching": caching,
}
def graph(
kind="PromptManager",
text=None,
role="positive",
sampler=True,
negative_text="",
seed=1,
):
"""PrimitiveString -> PreviewAny -> ``kind`` -> KSampler (runtime text).
``text`` set: typed text instead of the runtime link. ``role`` "negative":
the node feeds the sampler's negative input. Pair: its negative editor holds
``negative_text`` and it feeds both sampler inputs.
"""
g = {
"str": {"class_type": "PrimitiveString", "inputs": {"value": "seed text"}},
"llm": {"class_type": "PreviewAny", "inputs": {"source": ["str", 0]}},
"ckpt": {"class_type": "CheckpointLoaderSimple", "inputs": {"ckpt": "m"}},
"pm": {
"class_type": kind,
"inputs": {"text": ["llm", 0] if text is None else text},
},
"other": {
"class_type": "CLIPTextEncode",
"inputs": {"text": "", "clip": ["ckpt", 1]},
},
"save": {"class_type": "SaveImage", "inputs": {"images": ["ks", 0]}},
}
if kind != "PromptManagerText":
g["pm"]["inputs"]["clip"] = ["ckpt", 1]
if kind == "PromptManagerPair":
g["pm"]["inputs"]["negative_text"] = negative_text
source = ["pm", 0]
if kind == "PromptManagerText":
g["enc"] = {
"class_type": "CLIPTextEncode",
"inputs": {"text": ["pm", 0], "clip": ["ckpt", 1]},
}
source = ["enc", 0]
positive, negative = (source, ["other", 0])
if role == "negative":
positive, negative = negative, positive
if kind == "PromptManagerPair":
positive, negative = ["pm", 0], ["pm", 1]
if sampler:
g["ks"] = {
"class_type": "KSampler",
"inputs": {
"model": ["ckpt", 0],
"positive": positive,
"negative": negative,
"seed": seed,
},
}
else:
g["save"]["inputs"]["images"] = positive
return g
class JobCase(unittest.TestCase):
"""A database, the queue hook stashes and one ``CachedRuns`` per test."""
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.addCleanup(self.tmp.cleanup)
self.addCleanup(assert_closed, self, self.tmp.name)
self.db = PromptDatabase(os.path.join(self.tmp.name, "runs.db"))
self.addCleanup(self.db.close_all)
patcher = mock.patch.dict(sys.modules, host_modules())
patcher.start()
self.addCleanup(patcher.stop)
self.queued, self.pair_queued = QueuedRuns(), QueuedRuns()
self.pending, self.pair_pending = PendingFirstUse(), PendingFirstUse()
self.context = (None, None)
self.runs = CachedRuns(context=lambda: self.context)
self.queue = object()
self.executor = types.SimpleNamespace(
server=types.SimpleNamespace(prompt_queue=self.queue),
caches=types.SimpleNamespace(outputs=HierarchicalCache()),
)
self.item_id = 0
self.trackers = []
# -- nodes ----------------------------------------------------------------
def node(self, kind):
if kind == "PromptManagerPair":
from prompt_manager_pair import PromptManagerPair as cls
elif kind == "PromptManagerText":
from prompt_manager_text import PromptManagerText as cls
else:
import prompt_manager
cls = getattr(prompt_manager, kind)
node = cls.__new__(cls)
node.db = self.db
node.logger = logging.getLogger("test.cached_runs")
node.prompt_tracker = mock.Mock()
self.trackers.append(node.prompt_tracker)
node.comfyui_integration = mock.Mock()
node._inject_lora_trigger_words = lambda text: text
node.pending_first_use = self.pending
node.pending_pair_first_use = self.pair_pending
node.cached_runs = self.runs
return node
def execute(self, g, node_id="pm", text=LLM, negative="", index=None):
"""Run the node as ComfyUI would; returns its STRING outputs."""
kind = g[node_id]["class_type"]
node = self.node(kind)
self.context = (self.context[0], index)
clip = mock.Mock()
if kind == "PromptManagerText":
return node.process_text(text, prompt=g, unique_id=node_id)
if kind == "PromptManagerPair":
return node.encode_prompt(
clip, text, negative_text=negative, prompt=g, unique_id=node_id
)
return node.encode_prompt(clip, text, prompt=g, unique_id=node_id)
def cache(self, node_id, *results):
"""Store executed results (one per list index) as ComfyUI's cache entry."""
width = len(results[0])
outputs = [[result[slot] for result in results] for slot in range(width)]
self.executor.caches.outputs.entries[node_id] = Entry({}, outputs)
# -- jobs -----------------------------------------------------------------
def job(self, g, cached=(), run=None, outputs=("save",), prompt_id=None):
"""One queued job: hook, dequeue, ``execution_cached``, executions, done."""
self.item_id += 1
prompt_id = prompt_id or f"job-{self.item_id}"
request = handle_queued_prompt(
{"prompt": g, "prompt_id": prompt_id},
self.queued,
pair_queued=self.pair_queued,
)
on_dequeue(
request["prompt_id"],
self.queued,
lambda: self.db,
self.pending,
pair_queued=self.pair_queued,
pair_pending=self.pair_pending,
)
self.runs.begin_attempt(
((self.item_id, prompt_id, g, {}, list(outputs), {}), self.item_id)
)
self.context = (prompt_id, None)
self.runs.replay(self.executor, prompt_id, list(cached), lambda: self.db)
result = run() if run else None
self.runs.end_attempt(self.item_id)
self.context = (None, None)
return result
def first(self, g, text=LLM, negative="", node_id="pm"):
"""A job that executes the node and fills the cache."""
result = self.job(
g, run=lambda: self.execute(g, node_id, text=text, negative=negative)
)
self.cache(node_id, result)
return result
def count(self, text):
row = self.db.get_prompt_by_hash(generate_prompt_hash(text))
return row["run_count"] if row else None
def pair_count(self, positive, negative):
pair_id = find_pair_id(self.db, positive, negative)
return self.db.get_prompt_pair(pair_id)["run_count"] if pair_id else None
class TestCachedJobsCount(JobCase):
def test_runtime_text_counts_every_cached_job_for_each_node_kind(self):
for kind in (
"PromptManager",
"PromptManagerFocused",
"PromptManagerSplit",
"PromptManagerText",
):
with self.subTest(kind=kind):
text = f"{LLM} for {kind}"
g = graph(kind)
self.first(g, text=text)
self.job(graph(kind, seed=2), cached=["pm"])
self.job(graph(kind, seed=3), cached=["pm"])
self.assertEqual(self.count(text), 3)
def test_typed_text_still_counts_once_per_job(self):
g = graph(text="typed castle")
self.first(g, text="typed castle")
self.job(g, cached=["pm"])
self.job(g, cached=["pm"])
self.assertEqual(self.count("typed castle"), 3)
def test_rewired_to_negative_while_cached_counts_nothing(self):
self.first(graph())
self.job(graph(role="negative"), cached=["pm"])
self.assertEqual(self.count(LLM), 1)
# Rewired back to positive: the receipt still proves the text
self.job(graph(), cached=["pm"])
self.assertEqual(self.count(LLM), 2)
def test_no_sampler_counts_every_node_on_replay(self):
g = graph(sampler=False)
self.first(g)
self.job(g, cached=["pm"])
self.assertEqual(self.count(LLM), 2)
def test_empty_editor_replays_nothing(self):
g = graph()
self.first(g, text="")
self.job(g, cached=["pm"])
self.assertEqual(self.db.search_prompts(), [])
def test_alternating_upstream_values_each_count_their_own_prompt(self):
g = graph()
self.first(g, text="value A")
self.first(g, text="value B")
self.cache("pm", self.execute_outputs("value A"))
self.job(g, cached=["pm"])
self.assertEqual((self.count("value A"), self.count("value B")), (2, 1))
def execute_outputs(self, text):
return (mock.ANY, text)
def test_list_items_replay_each_item_in_order(self):
g = graph()
results = []
def run():
for index, text in enumerate(("one", "two", "three")):
results.append(self.execute(g, text=text, index=index))
self.job(g, run=run)
self.cache("pm", *results)
self.job(g, cached=["pm"])
self.assertEqual([self.count(t) for t in ("one", "two", "three")], [2, 2, 2])
def test_a_repeated_id_or_message_never_counts_a_node_twice(self):
g = graph()
self.first(g)
self.job(g, cached=["pm", "pm"])
self.assertEqual(self.count(LLM), 2)
self.item_id += 1
self.runs.begin_attempt(((1, "again", g, {}, ["save"], {}), self.item_id))
for _ in range(2):
self.runs.replay(self.executor, "again", ["pm"], lambda: self.db)
self.assertEqual(self.count(LLM), 3)
def test_a_shorter_list_keeps_its_receipt_after_a_longer_one(self):
g = graph()
self.first(g, text="alpha")
results = []
def run():
for index, text in enumerate(("alpha", "beta")):
results.append(self.execute(g, text=text, index=index))
self.job(g, run=run)
self.cache("pm", (mock.ANY, "alpha"))
self.job(g, cached=["pm"])
self.assertEqual((self.count("alpha"), self.count("beta")), (3, 1))
def test_a_job_reusing_a_prompt_id_counts_and_executes_normally(self):
g = graph()
self.first(g)
self.job(g, cached=["pm"], prompt_id="same")
self.job(g, run=lambda: self.execute(g), prompt_id="same")
self.job(g, cached=["pm"], prompt_id="same")
self.assertEqual(self.count(LLM), 4)
class TestFailClosed(JobCase):
def setUp(self):
super().setUp()
self.g = graph()
self.first(self.g)
def assertNotCounted(self):
self.assertEqual(self.count(LLM), 1)
def test_changed_cached_text_is_not_counted(self):
self.cache("pm", (mock.ANY, LLM + " edited"))
self.job(self.g, cached=["pm"])
self.assertNotCounted()
def test_no_receipt_in_this_process_is_not_counted(self):
self.runs = CachedRuns(context=lambda: self.context)
self.job(self.g, cached=["pm"])
self.assertNotCounted()
def test_missing_cache_entry_is_not_counted(self):
self.executor.caches.outputs.entries.clear()
self.job(self.g, cached=["pm"])
self.assertNotCounted()
def test_unsupported_cache_is_not_counted(self):
entries = self.executor.caches.outputs.entries
self.executor.caches.outputs = NullCache()
self.executor.caches.outputs.entries = entries
self.job(self.g, cached=["pm"])
self.assertNotCounted()
def test_more_list_values_than_the_bound_are_not_counted(self):
self.cache("pm", *[(mock.ANY, LLM)] * 65)
self.job(self.g, cached=["pm"])
self.assertNotCounted()
def test_missing_list_index_is_not_counted(self):
self.cache("pm", (mock.ANY, LLM), (mock.ANY, LLM))
self.job(self.g, cached=["pm"])
self.assertNotCounted()
def test_node_the_job_does_not_run_is_not_counted(self):
self.job(self.g, cached=["pm"], outputs=("other",))
self.assertNotCounted()
def test_message_for_another_job_is_not_counted(self):
self.item_id += 1
self.runs.begin_attempt(((1, "this", self.g, {}, ["save"], {}), self.item_id))
self.runs.replay(self.executor, "another", ["pm"], lambda: self.db)
self.assertNotCounted()
def test_unkeyable_node_is_not_counted(self):
g = graph()
g["pm"]["inputs"]["extra"] = ["missing", 0]
self.job(g, cached=["pm"])
self.assertIsNone(prompt_node_key(g, "pm"))
self.assertNotCounted()
def test_deleted_prompt_is_skipped(self):
self.db.delete_prompt(
self.db.get_prompt_by_hash(generate_prompt_hash(LLM))["id"]
)
self.job(self.g, cached=["pm"])
self.assertIsNone(self.count(LLM))
def test_database_errors_never_raise(self):
def broken():
raise RuntimeError("database unavailable")
self.item_id += 1
self.runs.begin_attempt(((1, "j", self.g, {}, ["save"], {}), self.item_id))
self.runs.replay(self.executor, "j", ["pm"], broken)
self.assertNotCounted()
def test_malformed_messages_and_results_are_ignored(self):
self.runs.begin_attempt(None)
self.runs.begin_attempt((("x",), "not an int"))
self.runs.replay(self.executor, None, ["pm"], lambda: self.db)
self.item_id += 1
self.runs.begin_attempt(((1, "j", self.g, {}, ["save"], {}), self.item_id))
self.runs.replay(self.executor, "j", "pm", lambda: self.db)
self.runs.replay(self.executor, "j", ["pm"] * 501, lambda: self.db)
self.runs.replay(self.executor, "j", [7, None], lambda: self.db)
self.assertNotCounted()
class TestReplayThenExecution(JobCase):
def test_execution_after_replay_in_the_same_job_is_not_counted_again(self):
# RAM-pressure frees the entry after execution_cached: the node executes
g = graph()
self.first(g)
self.job(g, cached=["pm"], run=lambda: self.execute(g))
self.assertEqual(self.count(LLM), 2)
# The suppressed execution still leaves a receipt: the next job counts
self.job(g, cached=["pm"])
self.assertEqual(self.count(LLM), 3)
def test_suppression_keeps_the_pending_catch_up_and_the_tracker_queue(self):
g = graph()
self.first(g)
prompt_id = self.db.get_prompt_by_hash(generate_prompt_hash(LLM))["id"]
def run():
# The replay counted, then the prompt was deleted and re-saved
self.db.delete_prompt(prompt_id)
self.pending.add(generate_prompt_hash(LLM))
return self.execute(g)
self.job(g, cached=["pm"], run=run)
self.assertEqual(self.count(LLM), 1) # the catch-up, not the own count
push = self.trackers[-1].set_current_prompt.call_args.kwargs["push_to_queue"]
self.assertTrue(push)
def test_without_host_context_the_node_counts_as_before(self):
g = graph()
self.runs = CachedRuns(context=lambda: (None, None))
self.first(g)
self.job(g, cached=["pm"])
self.job(g, run=lambda: self.execute(g))
self.assertEqual(self.count(LLM), 2)
class TestPairReplay(JobCase):
def test_runtime_positive_counts_positive_and_pair_every_job(self):
g = graph("PromptManagerPair", negative_text="blurry")
self.first(g, negative="blurry")
self.job(graph("PromptManagerPair", negative_text="blurry", seed=2), ["pm"])
self.job(graph("PromptManagerPair", negative_text="blurry", seed=3), ["pm"])
self.assertEqual(self.count(LLM), 3)
self.assertEqual(self.pair_count(LLM, "blurry"), 3)
self.assertEqual(self.count("blurry"), 0)
def test_empty_negative_pair_replays_its_pair_count(self):
g = graph("PromptManagerPair")
self.first(g)
self.job(g, cached=["pm"])
self.assertEqual(self.pair_count(LLM, ""), 2)
self.assertEqual(self.count(LLM), 2)
def test_pair_execution_after_replay_is_not_counted_again(self):
g = graph("PromptManagerPair", negative_text="blurry")
self.first(g, negative="blurry")
self.job(g, ["pm"], run=lambda: self.execute(g, negative="blurry"))
self.assertEqual(self.count(LLM), 2)
self.assertEqual(self.pair_count(LLM, "blurry"), 2)
def test_pair_role_follows_the_current_graph(self):
g = graph("PromptManagerPair", negative_text="blurry")
self.first(g, negative="blurry")
rewired = graph("PromptManagerPair", negative_text="blurry")
rewired["ks"]["inputs"]["positive"] = ["other", 0]
self.job(rewired, cached=["pm"])
self.assertEqual(self.count(LLM), 1)
self.assertEqual(self.pair_count(LLM, "blurry"), 1)
class TestOwnCounts(unittest.TestCase):
"""The node path and replay decide through the same function."""
def test_single_node_table(self):
facts = Facts("p")
for used, seen, expected in (
({"text"}, False, [("prompt", "p")]),
({"text"}, True, []),
(set(), False, []),
):
with self.subTest(used=used, seen=seen):
counts = own_counts(
"PromptManager", facts, Role(frozenset(used)), Seen(text=seen)
)
self.assertEqual(counts, expected)
self.assertEqual(
own_counts("PromptManager", Facts(None), Role(frozenset({"text"})), Seen()),
[],
)
def test_pair_table(self):
both = Role(frozenset({"text", "negative_text"}), True)
self.assertEqual(
own_counts("PromptManagerPair", Facts("p", "n"), both, Seen()),
[("prompt", "p"), ("prompt", "n"), ("pair", "p", "n")],
)
self.assertEqual(
own_counts("PromptManagerPair", Facts("p", "p"), both, Seen()),
[("prompt", "p"), ("pair", "p", "p")],
)
self.assertEqual(
own_counts("PromptManagerPair", Facts("p"), both, Seen(text=True)),
[("pair", "p", None)],
)
self.assertEqual(
own_counts(
"PromptManagerPair", Facts("p", "n"), both, Seen(True, True, True)
),
[],
)
def test_node_and_replay_agree_on_role_and_seen(self):
for g, node_id in (
(graph(), "pm"),
(graph(text="typed"), "pm"),
(graph(role="negative"), "pm"),
(graph(sampler=False), "pm"),
(graph("PromptManagerPair", negative_text="n"), "pm"),
(graph("PromptManagerPair", text="typed", negative_text="n"), "pm"),
):
kind = g[node_id]["class_type"]
text = "typed" if isinstance(g[node_id]["inputs"]["text"], str) else LLM
facts = stored_facts(text, g[node_id]["inputs"].get("negative_text"))
with self.subTest(kind=kind, inputs=g[node_id]["inputs"]):
replay = own_counts(
kind,
facts,
graph_role(g, node_id, kind),
graph_seen(g, node_id, kind, facts),
)
self.assertEqual(replay, self.node_counts(g, node_id, kind, text))
@staticmethod
def node_counts(g, node_id, kind, text):
"""What the executing node counted itself, from a recording database."""
db = mock.Mock()
db.get_prompt_by_hash.side_effect = lambda h: {"id": h}
db.save_prompt_pair.return_value = 1
negative = g[node_id]["inputs"].get("negative_text", "")
db.get_prompt_pair.return_value = {
"positive_prompt_id": generate_prompt_hash(text),
"negative_prompt_id": (
generate_prompt_hash(negative) if negative.strip() else None
),
}
case = JobCase()
case.db = db
case.pending, case.pair_pending = PendingFirstUse(), PendingFirstUse()
case.runs = CachedRuns(context=lambda: (None, None))
case.trackers = []
case.context = (None, None)
if kind == "PromptManagerPair":
node = case.node(kind)
node.encode_prompt(
mock.Mock(), text, negative_text=negative, prompt=g, unique_id=node_id
)
else:
node = case.node(kind)
node._track_prompt_execution(text, text, None, None, {}, g, node_id)
counts = [("prompt", c.args[0]) for c in db.record_prompt_use.call_args_list]
if kind == "PromptManagerPair":
counts += [
("pair", generate_prompt_hash(text), db.get_prompt_pair()[k])
for k in ["negative_prompt_id"]
if db.record_prompt_pair_use.called
]
return counts
class TestReceipts(unittest.TestCase):
def setUp(self):
self.context = ("j1", None)
self.runs = CachedRuns(
context=lambda: self.context, max_keys=2, max_receipts=2, max_items=3
)
self.g = {
name: {"class_type": "PromptManager", "inputs": {"text": name}}
for name in ("a", "b", "c")
}
def begin(self, item_id, prompt_id="j1"):
self.context = (prompt_id, None)
self.runs.begin_attempt(
((item_id, prompt_id, self.g, {}, ["a", "b", "c"], {}), item_id)
)
def keys(self):
return list(self.runs._receipts)
def test_least_recently_used_run_keys_are_dropped(self):
self.begin(1)
for name in ("a", "b", "c"):
self.runs.record_item(name, (name,), Facts("h"))
self.assertEqual(self.keys(), [prompt_node_key(self.g, n) for n in ("b", "c")])
def test_receipts_per_key_are_bounded_and_equal_outputs_replace(self):
key = prompt_node_key(self.g, "a")
for item_id, text in ((1, "x"), (2, "y"), (3, "z"), (4, "y")):
self.begin(item_id)
self.runs.record_item("a", (text,), Facts(text))
self.runs.end_attempt(item_id)
receipts = self.runs._receipts[key]
self.assertEqual([r.item_id for r in receipts], [4, 3])
def test_list_indices_past_the_bound_are_not_recorded(self):
self.begin(1)
self.context = ("j1", 3)
self.runs.record_item("a", ("x",), Facts("x"))
self.assertEqual(self.keys(), [])
def test_non_string_outputs_and_unknown_nodes_are_not_recorded(self):
self.begin(1)
self.runs.record_item("a", (None,), Facts("x"))
self.runs.record_item("zzz", ("x",), Facts("x"))
self.assertEqual(self.keys(), [])
def test_task_done_clears_only_its_own_attempt(self):
self.begin(1)
self.runs.end_attempt(2)
self.runs.record_item("a", ("x",), Facts("x"))
self.runs.end_attempt(1)
self.runs.record_item("b", ("x",), Facts("x"))
self.assertEqual(self.keys(), [prompt_node_key(self.g, "a")])
def test_job_nodes_follow_the_outputs_to_execute(self):
self.runs.begin_attempt(((1, "j1", self.g, {}, ["b"], {}), 1))
self.assertEqual(list(self.runs._attempt.nodes), ["b"])
class TestCacheReader(unittest.TestCase):
def setUp(self):
patcher = mock.patch.dict(sys.modules, host_modules())
patcher.start()
self.addCleanup(patcher.stop)
self.cache = LRUCache()
self.executor = types.SimpleNamespace(
caches=types.SimpleNamespace(outputs=self.cache)
)
def read(self, outputs, kind="PromptManager"):
self.cache.entries["n"] = Entry({}, outputs)
return cached_string_outputs(self.executor, "n", kind)
def test_reads_each_string_slot_in_order(self):
self.assertEqual(self.read([["c"], ["a", "b"]]), (("a", "b"),))
self.assertEqual(self.read([["t"]], "PromptManagerText"), (("t",),))
self.assertEqual(
self.read([[1], [2], ["p"], ["n"]], "PromptManagerPair"), (("p",), ("n",))
)
def test_fails_closed(self):
for outputs, kind in (
([["c"], ["a", 1]], "PromptManager"),
([["c"]], "PromptManager"),
([["c"], ["a"] * 65], "PromptManager"),
([[1], [2], ["p", "q"], ["n"]], "PromptManagerPair"),
([["c"], "text"], "PromptManager"),
([["c"], ["a"]], "CLIPTextEncode"),
([[]] * 5, "PromptManager"),
):
with self.subTest(outputs=outputs, kind=kind):
self.assertIsNone(self.read(outputs, kind))
self.cache.entries["m"] = ({}, [["c"], ["a"]]) # not a CacheEntry
self.assertIsNone(cached_string_outputs(self.executor, "m", "PromptManager"))
self.executor.caches.outputs = NullCache()
self.assertIsNone(self.read([["c"], ["a"]]))
class TestGraphHelpers(unittest.TestCase):
def test_upstream_nodes_and_closure(self):
g = graph()
self.assertEqual(
set(upstream_nodes(g, ["save"])),
{"save", "ks", "pm", "llm", "str", "ckpt", "other"},
)
self.assertEqual(upstream_nodes(g, ["missing"]), [])
self.assertEqual(upstream_nodes(None, ["save"]), [])
self.assertIsNone(upstream_closure(g, ["save"], limit=2))
self.assertIsNone(upstream_closure({"a": {"inputs": {"x": ["b", 0]}}}, ["a"]))
def test_find_pair_by_hashes_matches_the_text_lookup(self):
with tempfile.TemporaryDirectory() as tmp:
db = PromptDatabase(os.path.join(tmp, "p.db"))
try:
pair_id = db.save_prompt_pair("cat", "")
self.assertEqual(
find_pair_id_by_hashes(db, generate_prompt_hash("cat"), None),
pair_id,
)
self.assertIsNone(
find_pair_id_by_hashes(
db, generate_prompt_hash("cat"), generate_prompt_hash("x")
)
)
finally:
db.close_all()
assert_closed(self, tmp)
class TestRegistration(unittest.TestCase):
def setUp(self):
self.modules = host_modules()
patcher = mock.patch.dict(sys.modules, self.modules)
patcher.start()
self.addCleanup(patcher.stop)
flag = mock.patch.object(module, "_registered", threading.Event())
flag.start()
self.addCleanup(flag.stop)
class Queue:
def __init__(self):
self.done = []
def get(self, timeout=None):
return self.result
def task_done(self, item_id, *args, **kwargs):
self.done.append(item_id)
def test_registers_once_and_follows_the_job(self):
queue = self.Queue()
server = types.SimpleNamespace(prompt_queue=queue)
runs = mock.Mock()
self.assertTrue(register_cached_runs(server, lambda: "db", runs))
self.assertFalse(register_cached_runs(server, lambda: "db", runs))
queue.result = (("item",), 5)
queue.get()
runs.begin_attempt.assert_called_once_with((("item",), 5))
executor = self.modules["execution"].PromptExecutor()
executor.server = server
executor.add_message("execution_start", {"prompt_id": "j"}, broadcast=False)
executor.add_message(
"execution_cached", {"prompt_id": "j", "nodes": ["n"]}, broadcast=False
)
runs.replay.assert_called_once()
self.assertEqual(runs.replay.call_args.args[1:3], ("j", ["n"]))
other = self.modules["execution"].PromptExecutor()
other.server = types.SimpleNamespace(prompt_queue=object())
other.add_message("execution_cached", {"prompt_id": "j", "nodes": []}, False)
self.assertEqual(runs.replay.call_count, 1)
queue.task_done(5, None, status=None)
runs.end_attempt.assert_called_once_with(5)
def test_hosts_without_the_interfaces_register_nothing(self):
class NoTaskDone:
def get(self):
return None
self.assertFalse(
register_cached_runs(
types.SimpleNamespace(prompt_queue=NoTaskDone()), lambda: None
)
)
self.assertFalse(register_cached_runs(types.SimpleNamespace(), lambda: None))
del self.modules["execution"].PromptExecutor.add_message
self.assertFalse(
register_cached_runs(
types.SimpleNamespace(prompt_queue=self.Queue()), lambda: None
)
)
if __name__ == "__main__":
unittest.main()
+456
View File
@@ -0,0 +1,456 @@
"""The prompt-link read of the workspace Images page (#354).
``PromptDatabase.get_image_link_matches`` returns every row of its filter
SELECT in one SQLite step (one JSON value) instead of one step per row, so the
read does not have to win the GIL back once per linked image while ComfyUI
generates. Its result must equal the streamed read it replaces, which stays as
the fallback, on every filter and on odd values.
"""
import itertools
import os
import sqlite3
import sys
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from database import operations # noqa: E402
from database.operations import PromptDatabase, link_filter_query # noqa: E402
try:
from tests.open_handles import assert_closed # noqa: E402
except ImportError: # discovered with tests/ as the top-level directory
from open_handles import assert_closed # noqa: E402
LOGGER = "prompt_manager.database"
MISSING_FUNCTION_SQL = (
"SELECT pm_missing_function(json_group_array(json_array("
"file_path, rating, matched, positive))) FROM ({select})"
)
QUERIES = ("", "lantern", "LANTERN title", "no-such-text")
CATEGORIES = (None, "cat-a", "cat-b", "cat-none")
TAG_SETS = ([], ["t1"], ["t1", "t2"], ["t-none"])
PROMPTS = (
# id, text, category, rating, tags, library title
(1, "red lantern at dusk", "cat-a", 5, ("t1", "t2"), None),
(2, "blue harbor", "cat-a", None, ("t1",), None),
(3, "green field", "cat-b", 3, (), "Lantern title"),
(4, "pair positive", "cat-b", 4, ("t2",), None),
(5, "pair negative lantern", "cat-a", 1, ("t1",), None),
)
QUOTED = '/out/it\'s "quoted".png'
BACKSLASHED = "C:\\out\\back\\slash.png"
UNICODE = "/out/ünïcødé 画像.png"
NEWLINE = "/out/new\nline.png"
IMAGES = (
# prompt id, file path
(1, "/out/a.png"),
(2, QUOTED),
(2, BACKSLASHED),
(3, UNICODE),
(1, NEWLINE),
(1, "/out/ambiguous.png"),
(2, "/out/ambiguous.png"),
(5, "/out/pair.png"), # linked through pair 1: its positive is prompt 4
(3, "/out/two-pairs.png"), # two pairs: no single pair, so prompt 3
(1, None), # no file path: never listed
)
class LinkMatchesTestCase(unittest.TestCase):
def setUp(self):
self.temp = tempfile.TemporaryDirectory()
self.root = Path(os.path.realpath(self.temp.name))
self.db = PromptDatabase(str(self.root / "prompts.db"))
for name in ("_one_step_unsupported", "_one_step_warned"):
patcher = patch.object(operations, name, False)
patcher.start()
self.addCleanup(patcher.stop)
self.statements = []
original = self.db.model.get_connection
def traced_connection():
connection = original()
connection.set_trace_callback(self.statements.append)
return connection
patcher = patch.object(
self.db.model, "get_connection", side_effect=traced_connection
)
patcher.start()
self.addCleanup(patcher.stop)
def tearDown(self):
self.db.close_all()
assert_closed(self, self.root)
self.temp.cleanup()
def seed(self):
with self.db.model.get_connection() as conn:
for prompt_id, text, category, rating, tags, title in PROMPTS:
conn.execute(
"INSERT INTO prompts(id,text,category,rating,hash) "
"VALUES(?,?,?,?,?)",
(prompt_id, text, category, rating, f"hash-{prompt_id}"),
)
for tag in tags:
conn.execute("INSERT OR IGNORE INTO tags(name) VALUES(?)", (tag,))
conn.execute(
"INSERT INTO prompt_tags(prompt_id,tag_id) "
"SELECT ?, id FROM tags WHERE name=?",
(prompt_id, tag),
)
if title:
conn.execute(
"INSERT INTO prompt_library_state(prompt_id,title) "
"VALUES(?,?)",
(prompt_id, title),
)
ids = {}
for prompt_id, path in IMAGES:
cursor = conn.execute(
"INSERT INTO generated_images"
"(prompt_id,image_path,filename,file_path) VALUES(?,?,?,?)",
(prompt_id, path or "/elsewhere/x.png", "x.png", path),
)
ids.setdefault(path, cursor.lastrowid)
conn.execute(
"INSERT INTO prompt_pairs(id,positive_prompt_id,negative_prompt_id) "
"VALUES(1,4,5),(2,3,5),(3,3,NULL)"
)
conn.executemany(
"INSERT INTO image_prompt_pairs(image_id,pair_id,provenance) "
"VALUES(?,?,'{}')",
[
(ids["/out/pair.png"], 1),
(ids["/out/two-pairs.png"], 2),
(ids["/out/two-pairs.png"], 3),
],
)
def add_image(self, prompt_id, path):
with self.db.model.get_connection() as conn:
conn.execute(
"INSERT INTO generated_images(prompt_id,image_path,filename,file_path)"
" VALUES(?,?,?,?)",
(prompt_id, "/blob/x.png", "x.png", path),
)
def streamed(self, q, category, tags, every_link=False, wanted=None):
"""Today's read, called directly (the fallback)."""
sql, params = link_filter_query(
q, tags, every_link=every_link, category=category
)
with self.db.model.get_connection() as conn:
return operations._streamed_link_matches(
conn, sql, params, wanted, category is not None
)
def selects(self):
found = [s for s in self.statements if s.lstrip().upper().startswith("SELECT")]
self.statements.clear()
return found
class TestOneStepLinkRead(LinkMatchesTestCase):
def test_plain_read_returns_every_unambiguous_link(self):
self.seed()
matches = self.db.get_image_link_matches("", [], every_link=True)
self.assertEqual(
matches,
{
"/out/a.png": (True, 5, 1, None),
QUOTED: (True, None, 2, None),
BACKSLASHED: (True, None, 2, None),
UNICODE: (True, 3, 3, None),
NEWLINE: (True, 5, 1, None),
"/out/pair.png": (True, 4, 4, None),
"/out/two-pairs.png": (True, 3, 3, None),
},
)
self.assertIsNone(matches[QUOTED][1])
self.assertIsInstance(matches["/out/a.png"][1], int)
self.assertIsInstance(matches["/out/a.png"][2], int)
def test_search_hits_follow_text_title_and_tags(self):
self.seed()
self.assertEqual(
self.db.get_image_link_matches("lantern", []),
{
"/out/a.png": (True, 5, 1, None),
NEWLINE: (True, 5, 1, None),
UNICODE: (True, 3, 3, None),
"/out/two-pairs.png": (True, 3, 3, None),
},
)
self.assertEqual(
self.db.get_image_link_matches("lantern", [], category="cat-a"),
{
"/out/a.png": (True, 5, 1, "cat-a"),
NEWLINE: (True, 5, 1, "cat-a"),
QUOTED: (False, None, 2, "cat-a"),
BACKSLASHED: (False, None, 2, "cat-a"),
},
)
def test_every_filter_equals_the_streamed_read(self):
self.seed()
wanted_sets = (None, {"/out/a.png", UNICODE, "/out/pair.png", "/absent.png"})
for q, category, tags, every_link, wanted in itertools.product(
QUERIES, CATEGORIES, TAG_SETS, (False, True), wanted_sets
):
with self.subTest(
q=q, category=category, tags=tags, every=every_link, wanted=wanted
):
expected = self.streamed(q, category, tags, every_link, wanted)
self.statements.clear()
actual = self.db.get_image_link_matches(
q,
list(tags),
every_link=every_link,
wanted=wanted,
category=category,
)
self.assertEqual(actual, expected)
self.assertEqual(
{path: tuple(map(type, row)) for path, row in actual.items()},
{path: tuple(map(type, row)) for path, row in expected.items()},
)
found = self.selects()
self.assertEqual(len(found), 1, found)
self.assertIn("json_group_array", found[0])
def test_empty_database_returns_nothing_in_one_statement(self):
self.assertEqual(self.db.get_image_link_matches("", [], every_link=True), {})
self.assertEqual(self.db.get_image_link_matches("x", ["t"], category="c"), {})
found = self.selects()
self.assertEqual(len(found), 2, found)
self.assertTrue(all("json_group_array" in query for query in found))
def test_wanted_limits_the_result_to_the_listing(self):
self.seed()
self.assertEqual(
self.db.get_image_link_matches(
"", [], every_link=True, wanted={"/out/pair.png", "/x.png"}
),
{"/out/pair.png": (True, 4, 4, None)},
)
self.assertEqual(
self.db.get_image_link_matches("", [], every_link=True, wanted=set()),
{},
)
def test_category_read_carries_the_inherited_category_in_one_step(self):
# A recipe's own category wins over its positive's (#332 MC4); the
# read only narrows to links holding the value, the caller compares.
self.seed()
with self.db.model.get_connection() as conn:
conn.execute("UPDATE prompt_pairs SET category='cat-a' WHERE id=1")
self.statements.clear()
for category, expected in (
("cat-a", {"/out/pair.png": (True, 4, 4, "cat-a")}),
("cat-b", {"/out/pair.png": (True, 4, 4, "cat-a")}),
):
with self.subTest(category=category):
actual = self.db.get_image_link_matches(
"", [], every_link=True, category=category, wanted={"/out/pair.png"}
)
self.assertEqual(actual, expected)
self.assertEqual(
actual, self.streamed("", category, [], True, {"/out/pair.png"})
)
found = [query for query in self.selects() if "json_group_array" in query]
self.assertEqual(len(found), 2, found)
self.assertTrue(all("pair_category" in query for query in found))
def test_plain_read_leaves_the_categories_out_of_the_json(self):
self.seed()
self.statements.clear()
self.db.get_image_link_matches("", [], every_link=True)
(query,) = self.selects()
aggregate = query[query.index("json_array(") :].split(")", 1)[0]
self.assertEqual(aggregate, "json_array(file_path, rating, matched, positive")
def test_read_releases_its_snapshot(self):
# A statement left open after the read would pin its WAL snapshot, and
# a TRUNCATE checkpoint from another connection would report busy.
self.seed()
self.assertTrue(self.db.get_image_link_matches("", [], every_link=True))
other = sqlite3.connect(str(self.root / "prompts.db"))
try:
other.execute("UPDATE prompts SET rating=2 WHERE id=1")
other.commit()
busy, _log, _done = other.execute(
"PRAGMA wal_checkpoint(TRUNCATE)"
).fetchone()
finally:
other.close()
self.assertEqual(busy, 0)
self.assertEqual(os.path.getsize(self.root / "prompts.db-wal"), 0)
def test_wrapped_query_keeps_the_ambiguity_probe_an_index_search(self):
self.seed()
sql, params = link_filter_query(
"lantern", ["t1"], every_link=True, category="cat-a"
)
with self.db.model.get_connection() as conn:
plan = [
row[3]
for row in conn.execute(
"EXPLAIN QUERY PLAN "
+ operations.link_matches_json_query(sql, categories=True),
params,
)
]
self.assertIn(
"SEARCH other USING COVERING INDEX idx_image_file_path (file_path=?)", plan
)
self.assertFalse([step for step in plan if step.startswith("SCAN other")])
class TestStreamedFallback(LinkMatchesTestCase):
def test_missing_json_functions_stream_and_stop_trying(self):
self.seed()
expected = self.streamed("", None, [], True)
self.statements.clear()
wrapper = operations.link_matches_json_query
with (
patch.object(operations, "_LINK_MATCHES_JSON_SQL", MISSING_FUNCTION_SQL),
patch.object(
operations, "link_matches_json_query", side_effect=wrapper
) as attempts,
self.assertLogs(LOGGER, "WARNING") as logged,
):
first = self.db.get_image_link_matches("", [], every_link=True)
found = self.selects()
self.assertEqual(attempts.call_count, 1)
second = self.db.get_image_link_matches("", [], every_link=True)
again = self.selects()
self.assertEqual(first, expected)
self.assertEqual(second, expected)
self.assertEqual(len(logged.records), 1, logged.output)
self.assertIn("no such function: pm_missing_function", logged.output[0])
self.assertTrue(operations._one_step_unsupported)
# SQLite refuses the wrapped statement before it runs (not traced).
self.assertEqual(len(found), 1, found)
self.assertNotIn("json_group_array", found[0])
# The next read skips the attempt for the rest of the process.
self.assertEqual(attempts.call_count, 1)
self.assertEqual(len(again), 1, again)
self.assertNotIn("json_group_array", again[0])
def test_blob_path_streams_that_read_and_keeps_trying(self):
self.seed()
self.add_image(2, b"/out/blob.png")
expected = self.streamed("", None, [], True)
self.assertEqual(expected[b"/out/blob.png"], (True, None, 2, None))
self.statements.clear()
with self.assertLogs(LOGGER, "WARNING") as logged:
first = self.db.get_image_link_matches("", [], every_link=True)
second = self.db.get_image_link_matches("", [], every_link=True)
self.assertEqual(first, expected)
self.assertEqual(second, expected)
self.assertEqual(len(logged.records), 1, logged.output)
self.assertFalse(operations._one_step_unsupported)
found = self.selects()
self.assertEqual(sum("json_group_array" in query for query in found), 2)
self.assertEqual(len(found), 4, found)
def test_blob_that_reads_as_json_still_streams(self):
# SQLite 3.45+ reads a BLOB that is valid JSONB as JSON (b"\x00" is
# null) instead of refusing it; the streamed read keeps the bytes.
self.seed()
self.add_image(2, b"\x00")
expected = self.streamed("", None, [], True)
self.assertIn(b"\x00", expected)
with self.assertLogs(LOGGER, "WARNING"):
self.assertEqual(
self.db.get_image_link_matches("", [], every_link=True), expected
)
self.assertFalse(operations._one_step_unsupported)
def test_blob_rating_streams(self):
self.seed()
with self.db.model.get_connection() as conn:
# Only a database written around the CHECK can hold one.
conn.execute("PRAGMA ignore_check_constraints=ON")
conn.execute("UPDATE prompts SET rating=? WHERE id=3", (b"\x00",))
conn.execute("PRAGMA ignore_check_constraints=OFF")
expected = self.streamed("", None, [], True)
self.assertEqual(expected[UNICODE], (True, b"\x00", 3, None))
with self.assertLogs(LOGGER, "WARNING"):
self.assertEqual(
self.db.get_image_link_matches("", [], every_link=True), expected
)
def test_blob_category_streams_a_category_read(self):
self.seed()
with self.db.model.get_connection() as conn:
# Admitted by its recipe's text category; the positive's is a BLOB.
conn.execute("UPDATE prompts SET category=? WHERE id=4", (b"cat-b",))
conn.execute("UPDATE prompt_pairs SET category='cat-b' WHERE id=1")
expected = self.streamed("", "cat-b", [], True)
self.assertEqual(expected["/out/pair.png"], (True, 4, 4, "cat-b"))
with self.assertLogs(LOGGER, "WARNING") as logged:
actual = self.db.get_image_link_matches(
"", [], every_link=True, category="cat-b"
)
self.assertEqual(actual, expected)
self.assertIn("BLOB", logged.output[0])
self.assertFalse(operations._one_step_unsupported)
def test_json_text_over_the_length_limit_streams(self):
self.seed()
expected = self.streamed("", None, [], True)
conn = self.db.model.get_connection()
limit = conn.setlimit(sqlite3.SQLITE_LIMIT_LENGTH, 200)
try:
with self.assertLogs(LOGGER, "WARNING") as logged:
actual = self.db.get_image_link_matches("", [], every_link=True)
finally:
conn.setlimit(sqlite3.SQLITE_LIMIT_LENGTH, limit)
self.assertEqual(actual, expected)
self.assertIn("too big", logged.output[0])
self.assertFalse(operations._one_step_unsupported)
def test_an_interrupted_read_raises_without_a_fallback(self):
# Move library interrupts a retired connection: the read fails as
# before, and the one warning is kept for a real fallback.
self.seed()
conn = self.db.model.get_connection()
conn.set_progress_handler(lambda: 1, 1)
self.statements.clear()
try:
with (
self.assertNoLogs(LOGGER, "WARNING"),
self.assertRaisesRegex(sqlite3.OperationalError, "^interrupted$"),
):
self.db.get_image_link_matches("", [], every_link=True)
finally:
conn.set_progress_handler(None, 1)
self.assertFalse(operations._one_step_warned)
self.assertFalse(operations._one_step_unsupported)
self.assertFalse([s for s in self.statements if "json_group_array" not in s])
def test_the_maintenance_link_rows_reader_is_not_shadowed(self):
from database.maintenance import MaintenanceOperations
self.assertIs(PromptDatabase._link_rows, MaintenanceOperations._link_rows)
def test_a_failed_streamed_read_still_raises(self):
with (
patch.object(operations, "_LINK_MATCHES_JSON_SQL", MISSING_FUNCTION_SQL),
patch.object(operations, "_LINK_FILTER_SQL", "SELECT broken FROM nowhere"),
self.assertLogs(LOGGER, "WARNING"),
):
with self.assertRaises(sqlite3.OperationalError):
self.db.get_image_link_matches("", [], every_link=True)
if __name__ == "__main__":
unittest.main()
+277
View File
@@ -0,0 +1,277 @@
"""The inspector and History show an image's own category (#332 §5, DC4, DC5)."""
import os
import sqlite3
import unittest
from unittest.mock import patch
from tests import test_api_history_outputs as history
from tests.test_api_images import make_png
from py.api.images import _favorite_root_key, image_category_key
from utils.inspection_response import (
OWN_CATEGORY_UNREAD,
_library_projection,
apply_own_category,
)
class InspectCase(history.ImageAPITestCase):
async def tearDownAsync(self):
self.api.db.close_all()
await super().tearDownAsync()
def own(self, path, value, sources=None):
sources = sources or self.api._get_all_output_dirs()
self.api.db.set_image_categories([image_category_key(sources, path)], value)
async def library(self, path, root="0"):
response = await self.client.get(
f"/prompt_manager/images/inspect/{path}", params={"root": root}
)
body = await response.json()
self.assertEqual(response.status, 200, body)
return body["library"]
class TestInspectorOwnCategory(InspectCase):
async def test_inspect_reports_image_scope(self):
path = make_png(self.output_dir / "image.png")
self.own(path, "Portraits")
library = await self.library("image.png")
self.assertEqual(library["status"], "none")
self.assertEqual(
(library["category"], library["category_scope"]), ("Portraits", "image")
)
async def test_own_overrides_linked_recipe_and_ambiguous(self):
db = self.api.db
linked = make_png(self.output_dir / "linked.png")
self._link_image(linked, "castle at dusk")
prompt = db.get_prompt_id_for_image(str(linked))
with db.model.get_connection() as conn:
conn.execute(
"UPDATE prompts SET category='Landscapes' WHERE id=?", (prompt,)
)
library = await self.library("linked.png")
self.assertEqual(
(library["category"], library["category_scope"]),
("Landscapes", "positive"),
)
self.own(linked, "Portraits")
library = await self.library("linked.png")
self.assertEqual(library["status"], "linked")
self.assertEqual(
(library["category"], library["category_scope"]), ("Portraits", "image")
)
recipe = make_png(self.output_dir / "recipe.png")
self._save_prompt("Pair positive", category="Night")
pair = db.save_prompt_pair("Pair positive", "Noise", category="Recipes")
positive = db.get_prompt_pair(pair)["positive_prompt_id"]
image = db.link_image_to_prompt(positive, str(recipe))
db.link_image_to_pair(pair, image, {"source": "fixture"})
library = await self.library("recipe.png")
self.assertEqual(
(library["category"], library["category_scope"]), ("Recipes", "pair")
)
self.own(recipe, "Mine")
library = await self.library("recipe.png")
self.assertEqual(
(library["category"], library["category_scope"]), ("Mine", "image")
)
ambiguous = make_png(self.output_dir / "ambiguous.png")
self._link_image(ambiguous, "owner one")
self._link_image(ambiguous, "owner two")
self.own(ambiguous, "Scans")
library = await self.library("ambiguous.png")
self.assertEqual(library["status"], "ambiguous")
self.assertEqual(
(library["category"], library["category_scope"]), ("Scans", "image")
)
async def test_a_failed_read_hides_the_category(self):
linked = make_png(self.output_dir / "linked.png")
self._link_image(linked, "castle at dusk")
def broken(*_args, **_kwargs):
raise sqlite3.OperationalError("secret")
with patch.object(self.api.db, "image_category", side_effect=broken):
with self.assertLogs("prompt_manager", level="WARNING"):
library = await self.library("linked.png")
self.assertEqual(library["status"], "linked")
self.assertIsNone(library["category"])
self.assertIsNone(library["category_scope"])
async def test_a_lora_root_reads_no_own_category(self):
lora = self.comfy_dir / "loras"
make_png(lora / "preview.png")
roots = [self.output_dir, lora]
reads = []
real = self.api.db.image_category
def recorded(*args):
reads.append(args)
return real(*args)
with (
patch.object(self.api, "_media_roots", lambda outputs: roots),
patch.object(self.api, "_linked_only_error", self._no_error),
patch.object(self.api.db, "image_category", side_effect=recorded),
):
response = await self.client.get(
"/prompt_manager/images/inspect/preview.png", params={"root": "1"}
)
body = await response.json()
self.assertEqual(response.status, 200, body)
self.assertEqual(reads, [])
self.assertIsNone(body["library"]["category"])
@staticmethod
async def _no_error(*_args):
return None
class TestProjection(unittest.TestCase):
def test_scope_image_is_kept_for_category_only(self):
base = {"status": "none", "scope": "saved-library", "tags": []}
projected = _library_projection(
{**base, "category": "Scans", "category_scope": "image"}
)
self.assertEqual(projected["category_scope"], "image")
projected = _library_projection({**base, "tags_scope": "image"})
self.assertNotIn("tags_scope", projected)
def test_apply_own_category_three_cases(self):
library = {"status": "linked", "category": "L", "category_scope": "positive"}
self.assertEqual(
apply_own_category(library, "Mine"),
{"status": "linked", "category": "Mine", "category_scope": "image"},
)
self.assertEqual(apply_own_category(library, None), library)
self.assertEqual(
apply_own_category(library, OWN_CATEGORY_UNREAD),
{"status": "linked", "category": None, "category_scope": None},
)
self.assertEqual(library["category"], "L") # never mutated
class TestHistoryOwnCategory(history.ImageAPITestCase):
"""Only the inspection request reads the own category, keyed as Images."""
# The History fixture's helpers, without inheriting its tests.
_base = history.HistoryOutputAPITests
ref = _base.ref
capture = _base.capture
outputs = _base.outputs
current = _base.current
media = _base.media
async def get_application(self):
app = await super().get_application()
self.temp_dir = self.comfy_dir / "temp"
self.temp_dir.mkdir()
self.host_output = str(self.output_dir)
self.host_temp = str(self.temp_dir)
self.patches = history.ExitStack()
self.addCleanup(self.patches.close)
for name, value in (
("folder_paths.get_output_directory", lambda: self.host_output),
("folder_paths.get_temp_directory", lambda: self.host_temp),
):
self.patches.enter_context(patch(name, value, create=True))
self.patches.enter_context(
patch.object(
history.GalleryConfig,
"allowed_gallery_parents",
lambda: [str(self.comfy_dir)],
)
)
return app
async def tearDownAsync(self):
self.api.close_history_reads()
self.patches.close()
await super().tearDownAsync()
async def inspect(self, start):
descriptor = await self.media(start)
response = await self.current(descriptor["inspection_url"])
body = await response.json()
self.assertEqual(response.status, 200, body)
return body["library"]
async def test_history_inspection_shows_own_category(self):
path = make_png(self.output_dir / "image.png")
self.api.db.set_image_categories(
[image_category_key([self.output_dir], path)], "Portraits"
)
start = self.capture([self.ref()])
library = await self.inspect(start)
self.assertEqual(
(library["category"], library["category_scope"]), ("Portraits", "image")
)
async def test_nested_sources_give_the_images_key(self):
inner = self.output_dir / "nested"
path = make_png(inner / "image.png")
self.api._get_all_output_dirs = lambda: [self.output_dir, inner]
self.api.db.set_image_categories(
[(_favorite_root_key(inner), "image.png")], "S"
)
start = self.capture([self.ref(subfolder="nested")])
library = await self.inspect(start)
self.assertEqual(library["category"], "S")
self.assertEqual(
image_category_key([self.output_dir, inner], path),
(_favorite_root_key(inner), "image.png"),
)
async def test_a_symlinked_output_folder_still_shows_it(self):
real = self.comfy_dir / "real-output"
real.mkdir()
link = self.comfy_dir / "linked-output"
self._make_symlink(real, link, target_is_directory=True)
make_png(real / "image.png")
self.host_output = str(link)
self.api._get_all_output_dirs = lambda: [real]
self.api.db.set_image_categories([(_favorite_root_key(real), "image.png")], "S")
start = self.capture([self.ref()])
library = await self.inspect(start)
self.assertEqual(library["category"], "S")
async def test_outputs_and_media_read_no_own_category(self):
make_png(self.output_dir / "image.png")
start = self.capture([self.ref()])
reads = []
real = self.api.db.image_category
def recorded(*args):
reads.append(args)
return real(*args)
with patch.object(self.api.db, "image_category", side_effect=recorded):
descriptor = await self.media(start)
for field in ("url", "thumbnail_url"):
response = await self.current(descriptor[field])
await response.read()
self.assertEqual(response.status, 200)
self.assertEqual(reads, [])
await self.current(descriptor["inspection_url"])
self.assertEqual(len(reads), 1)
async def test_a_temp_output_has_none(self):
reference = {"folder_type": "temp", "subfolder": "", "filename": "a.png"}
self.assertIsNone(self.api._history_own_category(reference, None))
async def test_a_failed_read_hides_it(self):
make_png(self.output_dir / "image.png")
start = self.capture([self.ref()])
def broken(*_args):
raise sqlite3.OperationalError("secret")
with patch.object(self.api.db, "image_category", side_effect=broken):
with self.assertLogs("prompt_manager", level="WARNING"):
library = await self.inspect(start)
self.assertIsNone(library["category"])
self.assertTrue(os.path.exists(self.output_dir / "image.png"))
+29
View File
@@ -1426,3 +1426,32 @@ class TestPerRecordBudgets(unittest.TestCase):
self.assertEqual(metadata["record_formats"], {"workflow": "limited"})
self.assertEqual(metadata["workflow_graph"]["status"], "limited")
self.assertEqual(metadata["prompt_graph"]["status"], "absent")
class TestInheritedCategory(unittest.TestCase):
"""#332 MC4: one inherited-category rule for the chip, filter and History."""
def test_recipe_set_blank_and_absent(self):
from utils.inspection_response import inherited_category, library_metadata
cases = [
(("Night", "Recipes", True), ("Recipes", "pair")),
(("Night", " ", True), ("Night", "positive")),
(("Night", " \t", True), ("Night", "positive")),
(("Night", b" ", True), ("Night", "positive")),
(("Night", None, True), ("Night", "positive")),
(("Night", "Recipes", False), ("Night", "positive")),
((" ", None, False), (" ", "positive")),
((None, None, False), (None, "positive")),
]
for args, expected in cases:
with self.subTest(args=args):
self.assertEqual(inherited_category(*args), expected)
positive, recipe, has_recipe = args
metadata = library_metadata(
positive, ["t"], (recipe, []) if has_recipe else None
)
self.assertEqual(
(metadata["category"], metadata["category_scope"]), expected
)
self.assertEqual(metadata["tags_scope"], "positive")
+17
View File
@@ -85,6 +85,23 @@ class TestPairUsageTracking(unittest.TestCase):
self.db.get_prompt_by_id(pair["negative_prompt_id"])["run_count"], 0
)
def test_execution_records_a_cached_run_receipt_for_both_editors(self):
from utils.cached_runs import Facts
node = self.node()
node.cached_runs = Mock()
node.cached_runs.replayed.return_value = False
node.encode_prompt(
Mock(),
"cat",
negative_text="",
prompt=pair_graph(negative=""),
unique_id="1",
)
node.cached_runs.record_item.assert_called_once_with(
"1", ("cat", ""), Facts(usage_tracking.generate_prompt_hash("cat"), None)
)
def test_validation_rejection_never_counts_or_saves(self):
self.submit(pair_graph())
self.assertEqual(self.db.list_prompt_pairs()["total"], 0)
+44
View File
@@ -278,5 +278,49 @@ class TestNodeRecordsUse(UsageTestCase):
self.assertEqual(self.db.get_prompt_by_id(pid)["run_count"], 2)
class TestNodeRecordsCachedRunReceipt(UsageTestCase):
"""Each execution leaves a receipt for cached jobs (#347)."""
GRAPH = {
"1": {"class_type": "PromptManagerText", "inputs": {"text": "typed text"}},
}
def _node(self):
from unittest import mock
from prompt_manager_text import PromptManagerText
node = PromptManagerText.__new__(PromptManagerText)
node.db = self.db
node.logger = __import__("logging").getLogger("test.prompt_usage")
node.prompt_tracker = mock.Mock()
node.comfyui_integration = mock.Mock()
node._inject_lora_trigger_words = lambda text: text
node.cached_runs = mock.Mock()
node.cached_runs.replayed.return_value = False
return node
def test_receipt_holds_the_returned_text_and_the_stored_hash(self):
from utils.cached_runs import Facts
node = self._node()
node.process_text(
"typed text", prepend_text="pre", prompt=self.GRAPH, unique_id="1"
)
node.cached_runs.record_item.assert_called_once_with(
"1", ("pre typed text",), Facts(generate_prompt_hash("typed text"))
)
# Hook-visible text: the node counts nothing itself
row = self.db.get_prompt_by_hash(generate_prompt_hash("typed text"))
self.assertEqual(row["run_count"], 0)
def test_empty_editor_records_empty_facts(self):
from utils.cached_runs import Facts
node = self._node()
node.process_text(" ", prompt=self.GRAPH, unique_id="1")
node.cached_runs.record_item.assert_called_once_with("1", (" ",), Facts(None))
if __name__ == "__main__":
unittest.main()
+291 -1
View File
@@ -13,6 +13,11 @@ from database.library_commands import LibraryCommandError
from database.operations import PromptDatabase
from utils.hashing import generate_prompt_hash
try:
from test_image_categories import category_checksum
except ImportError:
from tests.test_image_categories import category_checksum
class TestVocabularyReads(unittest.TestCase):
def setUp(self):
@@ -168,7 +173,8 @@ class TestVocabularyReads(unittest.TestCase):
self.assertEqual(page["total"], 4)
self.assertFalse(page["has_more"])
for row in page["categories"]:
self.assertEqual(set(row), {"value", "prompt_count"})
self.assertEqual(set(row), {"value", "prompt_count", "image_count"})
self.assertEqual(row["image_count"], 0)
self.assertEqual(
row["prompt_count"],
self.db.list_library(category=row["value"])["total"],
@@ -238,6 +244,111 @@ class TestVocabularyReads(unittest.TestCase):
with self.assertRaises(ValueError):
self.read("list_vocabulary_tags", **options)
def own(self, value, *paths, root="root"):
"""Give each relative path under ``root`` the own category ``value``."""
self.db.set_image_categories([(root, path) for path in paths], value)
def all_categories(self, limit=200):
rows, pages, after = [], [], None
while True:
page = self.read("list_vocabulary_categories", limit=limit, after=after)
pages.append(page)
rows.extend(page["categories"])
if not page["has_more"]:
return rows, pages
after = page["next_key"]
def test_category_list_includes_image_only_values_last(self):
self.save("Oldest", category="Zeta")
self.save("Newer", category="Alpha")
self.own("Zeta", "a.png", "b.png")
self.own("Only images B", "c.png")
self.own("Only images A", "d.png", "e.png", "gone.png")
rows, _pages = self.all_categories()
self.assertEqual(
rows,
[
{"value": "Zeta", "prompt_count": 1, "image_count": 2},
{"value": "Alpha", "prompt_count": 1, "image_count": 0},
{"value": "Only images A", "prompt_count": 0, "image_count": 3},
{"value": "Only images B", "prompt_count": 0, "image_count": 1},
],
)
page = self.read("list_vocabulary_categories")
self.assertEqual(page["total"], 4)
def test_category_cursor_crosses_into_image_only_values(self):
for index in range(3):
self.save(f"Owner {index}", category=f"P{index}")
self.own("P1", "shared.png")
self.own("I-b", "x.png")
self.own("I-a", "y.png")
self.own("I-c", "z.png", root="other")
for limit in (1, 2, 3, 4):
with self.subTest(limit=limit):
rows, pages = self.all_categories(limit=limit)
self.assertEqual(
[row["value"] for row in rows],
["P0", "P1", "P2", "I-a", "I-b", "I-c"],
)
self.assertEqual({page["total"] for page in pages}, {6})
self.assertEqual(
[row["image_count"] for row in rows], [0, 1, 0, 1, 1, 1]
)
first = self.read("list_vocabulary_categories", limit=3)
self.assertEqual(first["next_key"], [3, "P2"])
second = self.read("list_vocabulary_categories", limit=1, after=[3, "P2"])
self.assertEqual(second["next_key"], [2**63 - 1, "I-a"])
def test_image_rows_on_a_prompt_value_count_once(self):
self.save("Owner", category="Shared")
self.db.save_prompt_pair("Owner", "Noise", category="Shared")
self.own("Shared", *[f"{n}.png" for n in range(5)])
self.own("Shared", "elsewhere.png", root="second")
rows, _pages = self.all_categories()
self.assertEqual(
rows, [{"value": "Shared", "prompt_count": 1, "image_count": 6}]
)
def test_library_options_ignore_image_categories(self):
self.save("Owner", category="Prompt value")
before = self.db.get_library_options()
self.own("Image value", "a.png")
self.own("Prompt value", "b.png")
self.assertEqual(self.db.get_library_options(), before)
self.assertEqual(before["categories"], ["Prompt value"])
def test_category_reads_use_the_image_category_index(self):
from database import vocabulary
self.own("Value", "a.png")
statements = [
vocabulary._CATEGORY_ROWS + "SELECT COUNT(*) FROM counted",
vocabulary._CATEGORY_ROWS
+ "SELECT value,first_pid,prompt_count,image_count FROM counted "
"WHERE first_pid>? OR (first_pid=? AND value>?) "
"ORDER BY first_pid,value LIMIT ?",
]
plans = []
with self.db.model.get_connection() as conn:
for sql in statements:
params = [1] * sql.count("?")
plans.extend(
row[-1] for row in conn.execute("EXPLAIN QUERY PLAN " + sql, params)
)
for sql in (
vocabulary._IMAGE_CATEGORY_ROWIDS,
*vocabulary._CATEGORY_UPDATES,
):
params = [1] * sql.count("?")
plans.extend(
row[-1] for row in conn.execute("EXPLAIN QUERY PLAN " + sql, params)
)
image_steps = [step for step in plans if "image_categories" in step]
self.assertTrue(image_steps)
for step in image_steps:
self.assertIn("idx_image_categories_category", step, step)
def test_reads_share_one_snapshot_and_report_epoch(self):
self.save("Cat", category="Animal", tags=["cat"])
epoch = self.db.get_execution_epoch()
@@ -741,6 +852,7 @@ class TestVocabularyCategoryCommands(unittest.TestCase):
"other_prompts": 1,
"negative_variants": 2,
"archived_negative_variants": 1,
"images": 0,
},
)
self.assertFalse(result["target_exists"])
@@ -810,6 +922,7 @@ class TestVocabularyCategoryCommands(unittest.TestCase):
"other_prompts": 0,
"negative_variants": 0,
"archived_negative_variants": 0,
"images": 0,
},
)
self.assertEqual(self.rows("prompts")[variant]["category"], "Other")
@@ -902,6 +1015,171 @@ class TestVocabularyCategoryCommands(unittest.TestCase):
self.apply(bad)
self.assertEqual(self.snapshot(), before)
def own(self, value, *paths, root="root"):
self.db.set_image_categories([(root, path) for path in paths], value)
def image_rows(self):
with self.db.model.get_connection() as conn:
return {
(row[0], row[1]): (row[2], row[3], row[4])
for row in conn.execute(
"SELECT root_key,relative_path,category,updated_at,rowid "
"FROM image_categories ORDER BY rowid"
)
}
def checksum(self):
with self.db.model.get_connection() as conn:
return category_checksum(conn)
def test_preview_counts_image_rows(self):
self.owners()
self.own("Scenery", "a.png", "b.png", "missing.png")
self.own("Scenery", "c.png", root="second")
self.own("Other", "d.png")
before = self.image_rows()
result = self.preview()
self.assertEqual(result["impact"]["images"], 4)
self.assertEqual(self.image_rows(), before)
self.assertTrue(self.preview(target="Other")["target_exists"])
def test_apply_renames_image_categories(self):
self.owners()
bystander = self.save("Bystander", category="Other")
self.own("Scenery", "a.png", "b.png")
self.own("Scenery", "c.png", root="second")
self.own(" Scenery ", "padded.png")
self.own("Other", "d.png")
self.age_rows()
with self.db.model.get_connection() as conn:
conn.execute("UPDATE image_categories SET updated_at=?", (_OLD_STAMP,))
images_before = self.image_rows()
prompts_before, pairs_before = self.snapshot()
preview = self.preview()
receipt = self.apply(preview["impact_token"])
self.assertEqual(receipt["impact"]["images"], 3)
images_after = self.image_rows()
self.assertEqual(set(images_after), set(images_before))
renamed = {("root", "a.png"), ("root", "b.png"), ("second", "c.png")}
for key, (value, stamp, rowid) in images_after.items():
old_value, old_stamp, old_rowid = images_before[key]
self.assertEqual(rowid, old_rowid, key)
if key in renamed:
self.assertEqual(value, "Landscape")
self.assertNotEqual(stamp, _OLD_STAMP)
else:
self.assertEqual((value, stamp), (old_value, old_stamp))
prompts_after, pairs_after = self.snapshot()
self.assertEqual(prompts_after[bystander], prompts_before[bystander])
for before, after in (
(prompts_before, prompts_after),
(pairs_before, pairs_after),
):
for row_id, row in after.items():
if before[row_id]["category"] != "Scenery":
self.assertEqual(row, before[row_id])
# Merging into a value images already hold keeps every file key.
merge = self.preview(source="Landscape", target="Other")
self.assertTrue(merge["target_exists"])
self.assertEqual(merge["impact"]["images"], 3)
receipt = self.apply(merge["impact_token"], source="Landscape", target="Other")
self.assertTrue(receipt["consolidated"])
merged = self.image_rows()
self.assertEqual(set(merged), set(images_before))
self.assertEqual(
sorted(key for key, row in merged.items() if row[0] == "Other"),
sorted([*renamed, ("root", "d.png")]),
)
def test_image_only_value_can_be_renamed(self):
self.save("Bystander", category="Kept")
self.db.save_prompt_pair("Pair", "Noise", category="Kept")
self.own("Image only", "a.png", "b.png")
self.age_rows()
before = self.checksum()
preview = self.preview(source="Image only", target="Renamed")
self.assertEqual(
preview["impact"],
{
"positive_prompts": 0,
"other_prompts": 0,
"negative_variants": 0,
"archived_negative_variants": 0,
"images": 2,
},
)
self.assertFalse(preview["target_exists"])
receipt = self.apply(
preview["impact_token"], source="Image only", target="Renamed"
)
self.assertFalse(receipt["consolidated"])
self.assertEqual({row[0] for row in self.image_rows().values()}, {"Renamed"})
self.assertEqual(self.checksum(), before)
# An image-only target counts as existing.
self.own("Image target", "c.png")
self.assertTrue(
self.preview(source="Renamed", target="Image target")["target_exists"]
)
# Nothing holds the value any more: refused.
with self.assertRaises(LibraryCommandError) as caught:
self.preview(source="Image only", target="Again")
self.assertEqual(caught.exception.code, "vocabulary-conflict")
def test_token_refused_after_an_image_row_gains_or_loses_the_value(self):
self.owners()
self.own("Scenery", "a.png", "b.png")
def gain():
self.own("Scenery", "late.png")
def lose():
self.db.clear_image_categories([("root", "a.png")])
def move_away():
self.own("Elsewhere", "b.png")
def target_on_image():
self.own("Landscape", "target.png")
for mutate in (gain, lose, move_away, target_on_image):
with self.subTest(mutate.__name__):
token = self.preview()["impact_token"]
mutate()
before = self.snapshot(), self.image_rows()
with self.assertRaises(LibraryCommandError) as caught:
self.apply(token)
self.assertEqual(caught.exception.code, "vocabulary-conflict")
self.assertEqual((self.snapshot(), self.image_rows()), before)
def test_image_category_failure_rolls_back_prompts(self):
self.owners()
self.own("Scenery", "a.png")
token = self.preview()["impact_token"]
with self.db.model.get_connection() as conn:
conn.execute(
"CREATE TRIGGER abort_images BEFORE UPDATE ON image_categories "
"BEGIN SELECT RAISE(ABORT, 'boom'); END"
)
before = self.snapshot(), self.image_rows()
with self.assertRaises(sqlite3.Error):
self.apply(token)
self.assertEqual((self.snapshot(), self.image_rows()), before)
def test_category_exists_reads_the_image_index(self):
from database import vocabulary
with self.db.model.get_connection() as conn:
plan = [
row[-1]
for row in conn.execute(
"EXPLAIN QUERY PLAN " + vocabulary._CATEGORY_EXISTS, {"v": "x"}
)
]
steps = [step for step in plan if "image_categories" in step]
self.assertTrue(steps, plan)
for step in steps:
self.assertIn("idx_image_categories_category", step)
def test_category_apply_failure_rolls_back(self):
self.owners()
token = self.preview()["impact_token"]
@@ -942,6 +1220,18 @@ class TestSharedTokenHashing(unittest.TestCase):
),
"cefe7f6a5544b7f6a203f5527810899c6af568aa7abc1b75adffbfb5ae6251ae",
)
with_images = category_impact_token(
"e", "S", "T", True, [(1, True)], [], images=[9, 4]
)
self.assertRegex(with_images, r"\A[0-9a-f]{64}\Z")
self.assertEqual(
with_images,
category_impact_token("e", "S", "T", True, [(1, True)], [], images=[4, 9]),
)
self.assertNotEqual(
with_images,
category_impact_token("e", "S", "T", True, [(1, True)], [], images=[4]),
)
if __name__ == "__main__":
+58 -2
View File
@@ -2,12 +2,14 @@ import test from 'node:test';
import assert from 'node:assert/strict';
import { createServer } from 'node:net';
import { spawn } from 'node:child_process';
import { once } from 'node:events';
import { EventEmitter, once } from 'node:events';
import { readFileSync } from 'node:fs';
import { isAbsolute } from 'node:path';
import { fileURLToPath, pathToFileURL } from 'node:url';
import { setTimeout as delay } from 'node:timers/promises';
import { startBrowserServer, createServerSupervisor, attachServerStderr, forwardPrefixedLines, SLOW_RESET_PREFIX } from '../../tools/vendor/start-browser-server.mjs';
import { workflowWorkers } from '../../tools/vendor/test-workers.mjs';
import { reporters, workflowWorkers } from '../../tools/vendor/test-workers.mjs';
import RunningTestsReporter from '../../tools/vendor/running-tests-reporter.mjs';
const fakeServer = fileURLToPath(new URL('./fake-browser-server.mjs', import.meta.url));
const fake = (...extra) => ({ command: process.execPath, args: port => [fakeServer, '--port', String(port), ...extra] });
@@ -255,3 +257,57 @@ test('worker count follows PM_TEST_WORKERS, then one in CI, then a quarter of th
assert.throws(() => workflowWorkers({ env: { PM_TEST_WORKERS: 'many' }, cpus: 8 }), /PM_TEST_WORKERS must be a positive integer/);
assert.throws(() => workflowWorkers({ env: { PM_TEST_WORKERS: '0' }, cpus: 8 }), /PM_TEST_WORKERS must be a positive integer/);
});
test('CI adds the running-tests reporter and a blob report under test-results/blob/, which the trace upload covers; local runs keep list', () => {
assert.equal(reporters({ env: {}, name: 'workflows' }), 'list');
const [list, running, blob] = reporters({ env: { CI: '1' }, name: 'workflows' });
assert.deepEqual(list, ['list']);
// Absolute: a config imported from another directory (workflow-artifacts.test.mjs) still finds it.
assert.ok(isAbsolute(running[0]) && running[0].endsWith('running-tests-reporter.mjs'), running[0]);
assert.deepEqual(blob, ['blob', { outputDir: './test-results/blob/workflows' }]);
});
function fakeHost() {
const host = new EventEmitter();
Object.assign(host, { pid: 4242, signals: [], cwd: () => '/repo/tools/vendor' });
host.kill = (pid, signal) => host.signals.push([pid, signal]);
return host;
}
const spec = (id, title, line) => ({ id, title, location: { file: '/repo/tests/browser/a.spec.mjs', line },
parent: { project: () => ({ name: 'firefox' }) } });
test('the running-tests reporter names only the tests that never ended, relative to rootDir', () => {
let clock = 1_000;
const output = [];
const host = fakeHost();
const reporter = new RunningTestsReporter({ now: () => clock, write: text => output.push(text), host });
reporter.onBegin({ rootDir: '/repo/tests/browser' });
const done = spec('1', 'finishes', 10);
const stuck = spec('2', 'hangs on a fetch', 20);
reporter.onTestBegin(done, { retry: 0 });
reporter.onTestBegin(stuck, { retry: 0 });
clock += 500;
reporter.onTestEnd(done, { retry: 0 });
clock += 41_500;
reporter.onEnd({ status: 'interrupted' });
assert.equal(output.join(''), '\nStill running when the run ended (interrupted): 1\n'
+ ' [firefox] › a.spec.mjs:20 › hangs on a fetch (running 42.0s)\n');
assert.equal(host.listenerCount('SIGTERM'), 0, 'the run is over: SIGTERM gets its default again');
output.length = 0;
reporter.onTestEnd(stuck, { retry: 0 });
reporter.onEnd({ status: 'passed' });
assert.deepEqual(output, [], 'a run where every test ended prints nothing');
});
test('on the first SIGTERM the reporter prints the running tests and hands over to Playwright\'s SIGINT', () => {
const output = [];
const host = fakeHost();
const reporter = new RunningTestsReporter({ now: () => 0, write: text => output.push(text), host });
reporter.onBegin({ rootDir: '/repo/tests/browser' });
reporter.onTestBegin(spec('2', 'hangs on a fetch', 20), { retry: 0 });
host.emit('SIGTERM');
assert.match(output.join(''), /Still running when the run ended \(SIGTERM\): 1\n {2}\[firefox\] › a\.spec\.mjs:20 › hangs on a fetch/);
assert.deepEqual(host.signals, [[4242, 'SIGINT']], 'Playwright\'s own interrupt ends the run and writes the blob');
assert.equal(host.listenerCount('SIGTERM'), 0, 'a second SIGTERM kills the process as usual');
});
+20
View File
@@ -247,6 +247,26 @@ test('browser job timeouts keep a real margin over measured runs', () => {
}
});
// A hang (#425) must end inside the test step, before the job limit, so the log and the always()
// uploads survive: the step's timeout-minutes < the job's.
const BROWSER_TEST_STEP = /npm run --prefix tools\/vendor test:(workflows|coverage)/;
test('each browser test step stops itself before its job limit and its traces upload anyway', () => {
for (const id of browserJobIds) {
const body = allJobs.get(id);
const jobMinutes = Number(/^ {4}timeout-minutes: (\d+)$/m.exec(body)[1]);
const jobSteps = steps(body);
const testAt = jobSteps.findIndex(step => BROWSER_TEST_STEP.test(step));
const testStep = jobSteps[testAt];
const stepMinutes = Number(/^ {8}timeout-minutes: (\d+)$/m.exec(testStep)?.[1]);
assert.ok(stepMinutes > 0, `${id}: the test step sets its own timeout-minutes`);
assert.ok(jobMinutes - stepMinutes >= 2, `${id}: the step limit leaves time for setup and the uploads`);
const upload = jobSteps.slice(testAt + 1).find(step => /^ {10}path: tools\/vendor\/test-results\/$/m.test(step));
assert.ok(upload, `${id} uploads tools/vendor/test-results/ (traces and blob report) after the tests`);
assert.match(upload, /^ {8}if: always\(\)$/m, `${id} uploads them when the step failed or timed out`);
}
});
// Every job is bounded (#350): a hang ends at a measured limit instead of GitHub's 6-hour default.
const GATING_WORKFLOWS = ['test.yml', 'code-quality.yml', 'viewerjs-check.yml'];
const BOUNDED_WORKFLOWS = [...GATING_WORKFLOWS, 'portability.yml'];
+4 -3
View File
@@ -34,15 +34,16 @@ test('execution script discovers only the new matrix and preserves all 78 origin
assert.match(executions, /\[classic-start\]/);
assert.match(executions, /\[nodes2-start\]/);
// + the fresh-host library location case, + #342 image attribution, + #343 linked text,
// + #332 image category routes on a generated output (image-attribution.spec.mjs, 2 projects)
assert.match(executions, /Total: 92 tests in 5 files/);
// + #332 image category routes on a generated output (image-attribution.spec.mjs, 2 projects),
// + #347 cached runtime-linked runs: five node kinds and eight one-renderer cases (execution-text-link.spec.mjs, 2 projects)
assert.match(executions, /Total: 118 tests in 5 files/);
for (const project of ['classic-start', 'nodes2-start']) {
const cases = executions.split('\n').filter(line => line.includes(`[${project}]`));
assert.equal(cases.filter(line => line.includes('execution-history.spec.mjs')).length, 15);
assert.equal(cases.filter(line => line.includes('execution-outputs.spec.mjs')).length, 8);
assert.equal(cases.filter(line => line.includes('library-location.spec.mjs')).length, 1);
assert.equal(cases.filter(line => line.includes('image-attribution.spec.mjs')).length, 3);
assert.equal(cases.filter(line => line.includes('execution-text-link.spec.mjs')).length, 19);
assert.equal(cases.filter(line => line.includes('execution-text-link.spec.mjs')).length, 32);
}
assert.doesNotMatch(executions, /prompt-manager.spec/);
});
+47 -2
View File
@@ -1,9 +1,11 @@
import test from 'node:test';
import assert from 'node:assert/strict';
import { existsSync, mkdtempSync, mkdirSync, rmSync, writeFileSync } from 'node:fs';
import { existsSync, mkdtempSync, mkdirSync, readFileSync, rmSync, writeFileSync } from 'node:fs';
import { EventEmitter } from 'node:events';
import { setTimeout as delay } from 'node:timers/promises';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { plan, runCoverage } from '../../tools/vendor/frontend-coverage.mjs';
import { plan, runCoverage, runNode } from '../../tools/vendor/frontend-coverage.mjs';
const steps = argv => plan(argv).map(({ step, ...options }) => (Object.keys(options).length ? { step, ...options } : step));
@@ -91,3 +93,46 @@ test('shard 1 runs the node units into the shared raw dir, then the viewer suite
assert.ok(calls[1].args.includes(join(paths.tools, 'playwright.config.mjs')));
assert.deepEqual(calls[2].args.slice(-2), ['--shard', '1/3']);
});
// #425: npm and a CI step timeout signal only frontend-coverage.mjs; its running child must get it.
function hostWithSignals() {
return new EventEmitter();
}
test('runNode resolves on exit 0, rejects with the exit code otherwise and leaves no signal listener', async () => {
const host = hostWithSignals();
await runNode(['-e', 'process.exit(0)'], {}, { host });
await assert.rejects(runNode(['-e', 'process.exit(3)'], {}, { host }), /Command failed \(3\): -e process\.exit\(3\)/);
assert.equal(host.listenerCount('SIGTERM') + host.listenerCount('SIGINT'), 0);
});
test('runNode forwards SIGTERM to the running child, which can report before it exits', async t => {
const root = mkdtempSync(join(tmpdir(), 'pm-coverage-signal-'));
t.after(() => rmSync(root, { recursive: true, force: true }));
const ready = join(root, 'ready');
const seen = join(root, 'seen');
const child = `const fs = require('node:fs');
process.on('SIGTERM', () => { fs.writeFileSync(${JSON.stringify(seen)}, 'SIGTERM'); process.exit(7); });
fs.writeFileSync(${JSON.stringify(ready)}, ''); setInterval(() => {}, 1000);`;
const host = hostWithSignals();
const running = runNode(['-e', child], {}, { host });
while (!existsSync(ready)) await delay(20);
host.emit('SIGTERM');
await assert.rejects(running, /Command failed \(7\)/);
assert.equal(readFileSync(seen, 'utf8'), 'SIGTERM', 'the child handled the forwarded signal');
assert.equal(host.listenerCount('SIGTERM'), 0);
});
test('runNode forwards SIGINT and starts the child in its own process group', async () => {
const host = hostWithSignals();
const kills = [];
let options;
const fake = new EventEmitter();
fake.kill = signal => { kills.push(signal); queueMicrotask(() => fake.emit('exit', null, signal)); };
const running = runNode(['x.mjs'], {}, { host, spawnChild: (command, args, spawnOptions) => { options = spawnOptions; return fake; } });
assert.equal(options.detached, true, 'a group-wide signal reaches the child only through forwarding, once');
host.emit('SIGINT');
await assert.rejects(running, /Command failed \(null\): x\.mjs/);
assert.deepEqual(kills, ['SIGINT']);
assert.equal(host.listenerCount('SIGINT'), 0);
});
+26 -9
View File
@@ -1,7 +1,7 @@
import { existsSync, readFileSync, readdirSync, mkdirSync, rmSync, writeFileSync } from 'node:fs';
import { dirname, join, relative } from 'node:path';
import { fileURLToPath, pathToFileURL } from 'node:url';
import { spawnSync } from 'node:child_process';
import { spawn } from 'node:child_process';
import { repositoryRoot } from './vendor-viewerjs.mjs';
import { discoverSources, checkCoverage } from '@comfyassets/toolkit/testing/coverage';
@@ -10,10 +10,27 @@ const metrics = ['lines', 'branches', 'functions', 'statements'];
const SHARD = /^([1-9]\d*)\/([1-9]\d*)$/;
const USAGE = 'usage: frontend-coverage.mjs [--update-baseline] | --shard I/N | --merge-only [--update-baseline]';
function runNode(args, env = {}) {
const result = spawnSync(process.execPath, args, { cwd: repositoryRoot, stdio: 'inherit',
env: { ...process.env, ...env } });
if (result.status !== 0) throw new Error(`Command failed (${result.status}): ${args.join(' ')}`);
const FORWARDED_SIGNALS = ['SIGTERM', 'SIGINT'];
// Runs one step's node child and waits for it. npm and a CI step timeout signal only this process,
// so SIGTERM and SIGINT are forwarded to the running child (#425): Playwright then reports the tests
// still running and writes its blob. The child gets its own process group, so a signal sent to the
// whole group (Ctrl-C, `timeout`) still reaches it exactly once, through this forwarding.
export function runNode(args, env = {}, { spawnChild = spawn, host = process } = {}) {
return new Promise((resolve, reject) => {
const child = spawnChild(process.execPath, args, { cwd: repositoryRoot, stdio: 'inherit',
env: { ...process.env, ...env }, detached: true });
const forward = signal => () => child.kill(signal);
const handlers = FORWARDED_SIGNALS.map(signal => [signal, forward(signal)]);
for (const [signal, handler] of handlers) host.on(signal, handler);
const done = () => { for (const [signal, handler] of handlers) host.removeListener(signal, handler); };
child.once('error', error => { done(); reject(error); });
child.once('exit', code => {
done();
if (code === 0) resolve();
else reject(new Error(`Command failed (${code}): ${args.join(' ')}`));
});
});
}
function defaultPaths() {
@@ -112,17 +129,17 @@ export async function runCoverage({ argv = process.argv.slice(2), run = runNode,
'node-units'() {
const tests = readdirSync(join(root, 'tests/js')).filter(file => file.endsWith('.test.js'))
.map(file => `tests/js/${file}`);
run(['--test', ...tests], { NODE_V8_COVERAGE: raw });
return run(['--test', ...tests], { NODE_V8_COVERAGE: raw });
},
viewer() {
run([playwright, 'test', '--config', join(tools, 'playwright.config.mjs')]);
return run([playwright, 'test', '--config', join(tools, 'playwright.config.mjs')]);
},
workflows({ shard }) {
run([playwright, 'test', '--config', join(tools, 'playwright.workflows.config.mjs'),
return run([playwright, 'test', '--config', join(tools, 'playwright.workflows.config.mjs'),
'--project', 'chromium', '--project', 'touch', ...(shard ? ['--shard', shard] : [])], { PM_JS_COVERAGE: raw });
},
report() {
run([join(tools, 'node_modules/c8/bin/c8.js'), 'report', '--all', '--src', sourceRoot,
return run([join(tools, 'node_modules/c8/bin/c8.js'), 'report', '--all', '--src', sourceRoot,
'--include', 'web/js/**/*.js', '--include', 'web/js/**/*.mjs', '--temp-directory', raw, '--reports-dir', reportRoot,
'--reporter', 'text', '--reporter', 'json-summary', '--reporter', 'html']);
},
+2 -1
View File
@@ -1,4 +1,5 @@
import { defineConfig } from '@playwright/test';
import { reporters } from './test-workers.mjs';
export default defineConfig({
testDir: './tests',
@@ -8,7 +9,7 @@ export default defineConfig({
workers: 2,
forbidOnly: Boolean(process.env.CI),
retries: process.env.CI ? 1 : 0,
reporter: 'list',
reporter: reporters({ name: 'viewer' }),
use: { browserName: 'chromium', headless: true, trace: 'retain-on-failure' },
projects: [
{ name: 'desktop', use: { viewport: { width: 1280, height: 800 } } },
+2 -2
View File
@@ -1,5 +1,5 @@
import { defineConfig, devices } from '@playwright/test';
import { workflowWorkers } from './test-workers.mjs';
import { reporters, workflowWorkers } from './test-workers.mjs';
// No global webServer: tests/browser/fixtures.mjs starts one server per worker
// on an OS-chosen port. Files run in parallel; tests inside a file keep their order.
@@ -11,7 +11,7 @@ export default defineConfig({
fullyParallel: false,
forbidOnly: Boolean(process.env.CI),
retries: 0,
reporter: 'list',
reporter: reporters({ name: 'workflows' }),
use: { trace: 'retain-on-failure', screenshot: 'only-on-failure' },
projects: [
{ name: 'chromium', outputDir: './test-results/workflows/chromium', use: { browserName: 'chromium' } },
+58
View File
@@ -0,0 +1,58 @@
import { writeSync } from 'node:fs';
import { relative } from 'node:path';
// Names the tests still running when a run is cut off (#425). A browser test step that reaches its
// timeout-minutes gets SIGTERM, which ends Playwright at once: no reporter's onEnd runs, so neither
// `list` nor the blob report says what was running. On the first SIGTERM this reporter prints every
// started test that has not ended, then turns the signal into SIGINT, Playwright's own interrupt,
// which stops the workers, ends the run as interrupted (non-zero exit) and writes the blob report.
// A second SIGTERM gets the default behaviour: the process dies.
export default class RunningTestsReporter {
constructor({ now = () => Date.now(), write = text => writeSync(1, text), host = process } = {}) {
this.now = now;
this.write = write;
this.host = host;
this.rootDir = host.cwd();
this.running = new Map();
this.onSigterm = () => {
this.host.removeListener('SIGTERM', this.onSigterm);
this.report('SIGTERM');
this.host.kill(this.host.pid, 'SIGINT');
};
}
printsToStdio() {
return false;
}
onBegin(config) {
this.rootDir = config.rootDir;
this.host.on('SIGTERM', this.onSigterm);
}
onTestBegin(test, result) {
this.running.set(`${test.id}#${result.retry}`, { test, startedAt: this.now() });
}
onTestEnd(test, result) {
this.running.delete(`${test.id}#${result.retry}`);
}
describe({ test, startedAt }) {
const project = test.parent?.project()?.name;
const where = `${relative(this.rootDir, test.location.file)}:${test.location.line}`;
const seconds = ((this.now() - startedAt) / 1000).toFixed(1);
return `${project ? `[${project}] › ` : ''}${where} › ${test.title} (running ${seconds}s)`;
}
report(reason) {
if (!this.running.size) return;
const lines = [...this.running.values()].map(entry => ` ${this.describe(entry)}`);
this.write(`\nStill running when the run ended (${reason}): ${this.running.size}\n${lines.join('\n')}\n`);
}
onEnd(result) {
this.host.removeListener('SIGTERM', this.onSigterm);
this.report(result.status);
}
}
+11
View File
@@ -1,4 +1,8 @@
import { availableParallelism } from 'node:os';
import { fileURLToPath } from 'node:url';
// Absolute, so a config that imports this one from another directory still finds it.
const RUNNING_TESTS_REPORTER = fileURLToPath(new URL('./running-tests-reporter.mjs', import.meta.url));
const MAX_LOCAL_WORKERS = 8;
const CPUS_PER_WORKER = 4;
@@ -16,3 +20,10 @@ export function workflowWorkers({ env = process.env, cpus = availableParallelism
if (env.CI) return 1;
return Math.max(1, Math.min(MAX_LOCAL_WORKERS, Math.floor(cpus / CPUS_PER_WORKER)));
}
// Under CI, running-tests-reporter.mjs names the tests a step timeout's signal cut off (`list`
// cannot: they never end), and a blob report beside the traces is uploaded with them.
export function reporters({ env = process.env, name } = {}) {
if (!env.CI) return 'list';
return [['list'], [RUNNING_TESTS_REPORTER], ['blob', { outputDir: `./test-results/blob/${name}` }]];
}
+532
View File
@@ -0,0 +1,532 @@
"""Count a cached PromptManager node like the execution it reuses (#347).
The queue hook (``usage_tracking``) counts text it can read from the queued
graph. Text only known at run time (a PreviewAny or LLM link, a list node) is
counted by the node itself when it executes. When ComfyUI later serves the node
from its output cache the node does not execute, so without this module that
job is not counted.
When a PromptManager-family node executes it records a *receipt item*: the
role-independent facts its own count is made from (the stored text hashes)
and digests of the STRING outputs it returned. When ComfyUI reports the node
as cached for a later job (``execution_cached``), the receipt whose digests
equal the text in the cache entry is replayed once for that job. The node's
role (positive, negative, which Pair outputs are used) and what the hook
already saw are recomputed from that job's own graph, and both paths count
through the same ``own_counts`` so they cannot drift.
Fail closed: when the cached text cannot be matched to a receipt recorded in
this process, nothing is counted (the behavior before #347). In memory only and
bounded; starts no threads.
"""
import hashlib
import threading
from collections import OrderedDict
from typing import Any, Callable, Dict, FrozenSet, List, NamedTuple, Optional, Tuple
try:
from .hashing import generate_prompt_hash
from .logging_config import get_logger
from .prompt_graph import (
PROMPT_MANAGER_TYPES,
intrinsic_pair_nodes,
resolve_text,
run_prompt_components,
run_prompt_nodes,
)
from .prompt_pair_resolver import resolve_text_input
from .queue_observers import subscribe_dequeue, subscribe_task_done
from .run_keys import prompt_node_key, upstream_nodes
from .usage_tracking import find_pair_id_by_hashes, survive_switch
except ImportError:
from utils.hashing import generate_prompt_hash
from utils.logging_config import get_logger
from utils.prompt_graph import (
PROMPT_MANAGER_TYPES,
intrinsic_pair_nodes,
resolve_text,
run_prompt_components,
run_prompt_nodes,
)
from utils.prompt_pair_resolver import resolve_text_input
from utils.queue_observers import subscribe_dequeue, subscribe_task_done
from utils.run_keys import prompt_node_key, upstream_nodes
from utils.usage_tracking import find_pair_id_by_hashes, survive_switch
logger = get_logger("prompt_manager.cached_runs")
PAIR = "PromptManagerPair"
MAX_KEYS = 256 # run keys remembered (the same bound as MAX_KEYED_RUNS)
MAX_RECEIPTS = 4 # receipts per run key: a recurring upstream value (A, B, A)
MAX_ITEMS = 64 # list items per receipt (the cache reader's bound)
MAX_CACHED_NODES = 500 # the execution_cached guard the history capture uses
_EMPTY_HASH = generate_prompt_hash("")
Count = Tuple[Any, ...] # ("prompt", hash) or ("pair", positive, negative or None)
class Facts(NamedTuple):
"""What one execution stored: hashes of the stored text, None when nothing.
``positive_hash`` is None for an empty editor (or saving disabled);
``negative_hash`` (Pair only) is None for an empty negative, which the pair
stores as ``negative_prompt_id IS NULL``.
"""
positive_hash: Optional[str]
negative_hash: Optional[str] = None
class Role(NamedTuple):
"""How a job's graph uses the node: which editors reach a sampler's positive."""
used_fields: FrozenSet[str]
pair_used: bool = False
class Seen(NamedTuple):
"""Which of the node's texts the queue hook resolved, and so already counted."""
text: bool = False
negative_text: bool = False
pair: bool = False
class _Item(NamedTuple):
outputs: Tuple[str, ...]
facts: Facts
class _Receipt:
__slots__ = ("item_id", "items")
def __init__(self, item_id: int) -> None:
self.item_id = item_id
self.items: Dict[int, _Item] = {}
def outputs(self) -> Dict[int, Tuple[str, ...]]:
return {index: item.outputs for index, item in self.items.items()}
class _Attempt(NamedTuple):
item_id: int
prompt_id: str
graph: Dict[str, Any]
nodes: Dict[str, Tuple[str, str]] # node id -> (class, run key)
replayed: set
def text_digest(value: str) -> str:
"""Digest of one returned STRING value, exactly as returned."""
return hashlib.sha256(value.encode("utf-8", "surrogatepass")).hexdigest()
def stored_facts(text: Any, negative_text: Any = None) -> Facts:
"""The facts of an execution that stores ``text`` (and a Pair's negative)."""
positive = text if isinstance(text, str) and text.strip() else None
negative = (
negative_text
if isinstance(negative_text, str) and negative_text.strip()
else None
)
return Facts(
generate_prompt_hash(positive) if positive is not None else None,
generate_prompt_hash(negative) if negative is not None else None,
)
def pair_components(facts: Facts, role: Role) -> List[Tuple[str, str]]:
"""``(field, hash)`` of each used, stored Pair component; each hash once."""
found, counted = [], set()
for field, prompt_hash in (
("text", facts.positive_hash),
("negative_text", facts.negative_hash),
):
if field not in role.used_fields or prompt_hash is None:
continue
if prompt_hash in counted:
continue
counted.add(prompt_hash)
found.append((field, prompt_hash))
return found
def own_counts(node_type: str, facts: Facts, role: Role, seen: Seen) -> List[Count]:
"""What one execution of a node counts itself: text the hook could not see.
Shared by the executing node and by replay. Excludes the pending first-use
catch-up, which only an execution that saves the prompt can consume.
"""
if facts.positive_hash is None:
return []
if node_type != PAIR:
if "text" in role.used_fields and not seen.text:
return [("prompt", facts.positive_hash)]
return []
counts: List[Count] = [
("prompt", prompt_hash)
for field, prompt_hash in pair_components(facts, role)
if not getattr(seen, field)
]
if role.pair_used and not seen.pair:
counts.append(("pair", facts.positive_hash, facts.negative_hash))
return counts
def graph_role(graph: Any, node_id: str, node_type: str) -> Role:
"""The node's role in a queued graph, by the functions the nodes use."""
if node_type != PAIR:
used = {"text"} if node_id in run_prompt_nodes(graph) else set()
return Role(frozenset(used))
used = {
field for source, field in run_prompt_components(graph) if source == node_id
}
return Role(frozenset(used), node_id in intrinsic_pair_nodes(graph))
def _same(resolved: Optional[str], prompt_hash: Optional[str]) -> bool:
return resolved is not None and generate_prompt_hash(resolved) == prompt_hash
def graph_seen(graph: Any, node_id: str, node_type: str, facts: Facts) -> Seen:
"""Which texts the queue hook resolved from ``graph`` (as the node decides)."""
if node_type != PAIR:
return Seen(text=_same(resolve_text(graph, node_id), facts.positive_hash))
positive = _same(resolve_text_input(graph, node_id, "text"), facts.positive_hash)
negative = _same(
resolve_text_input(graph, node_id, "negative_text"),
facts.negative_hash or _EMPTY_HASH,
)
return Seen(positive, negative, positive and negative)
def _host_context() -> Tuple[Optional[str], Optional[int]]:
"""(prompt id, list index) of the node executing on this thread, if known."""
try:
from comfy_execution.utils import get_executing_context
except Exception:
return None, None
try:
context = get_executing_context()
except Exception:
return None, None
prompt_id = getattr(context, "prompt_id", None)
index = getattr(context, "list_index", None)
return (
prompt_id if isinstance(prompt_id, str) and prompt_id else None,
index if type(index) is int else None,
)
def write_count(db: Any, count: Count) -> bool:
"""Record one counted run; False when the prompt or pair no longer exists."""
def write() -> bool:
if count[0] == "prompt":
row = db.get_prompt_by_hash(count[1])
if row:
db.record_prompt_use(row["id"])
return bool(row)
pair_id = find_pair_id_by_hashes(db, count[1], count[2])
if pair_id:
db.record_prompt_pair_use(pair_id)
return bool(pair_id)
return bool(survive_switch(db, write))
class CachedRuns:
"""Receipts of executed nodes and the one job currently executing."""
def __init__(
self,
*,
max_keys: int = MAX_KEYS,
max_receipts: int = MAX_RECEIPTS,
max_items: int = MAX_ITEMS,
context: Callable[[], Tuple[Optional[str], Optional[int]]] = _host_context,
) -> None:
self._max_keys = max_keys
self._max_receipts = max_receipts
self._max_items = max_items
self._context = context
self._receipts: "OrderedDict[str, List[_Receipt]]" = OrderedDict()
self._attempt: Optional[_Attempt] = None
self._lock = threading.Lock()
# -- the job ------------------------------------------------------------
def begin_attempt(self, result: Any) -> None:
"""A job was dequeued: ``result`` is ``PromptQueue.get``'s return value."""
attempt = None
try:
item, item_id = result[0], result[1]
prompt_id, graph, outputs = item[1], item[2], item[4]
if type(item_id) is int and isinstance(prompt_id, str):
if isinstance(graph, dict):
attempt = _Attempt(
item_id, prompt_id, graph, _job_nodes(graph, outputs), set()
)
except Exception as error:
logger.debug(f"Dequeued job not tracked for cached runs: {error}")
with self._lock:
self._attempt = attempt
def end_attempt(self, item_id: Any) -> None:
"""The job finished: its receipts are complete and replace equal older ones.
Deduplicating only complete receipts keeps a list execution that is
still recording ([A] so far of [A, B]) from dropping an older [A].
"""
with self._lock:
if self._attempt is not None and self._attempt.item_id == item_id:
self._attempt = None
for run_key, receipts in self._receipts.items():
if receipts and receipts[0].item_id == item_id:
self._receipts[run_key] = _deduplicated(receipts)
def _current(self) -> Tuple[Optional[_Attempt], Optional[int]]:
"""The attempt the executing node belongs to, and its list index."""
prompt_id, index = self._context()
with self._lock:
attempt = self._attempt
if attempt is None or prompt_id is None or prompt_id != attempt.prompt_id:
return None, None
return attempt, index
# -- node execution -----------------------------------------------------
def replayed(self, node_id: Any) -> bool:
"""Whether this job already counted the executing node by replay."""
attempt, _ = self._current()
if attempt is None:
return False
with self._lock:
return str(node_id) in attempt.replayed
def record_item(self, node_id: Any, outputs: Tuple[Any, ...], facts: Facts) -> None:
"""Remember what this execution returned and was entitled to count."""
try:
attempt, index = self._current()
node = attempt.nodes.get(str(node_id)) if attempt else None
if node is None:
return
index = 0 if index is None else index
if not 0 <= index < self._max_items:
return
if any(not isinstance(value, str) for value in outputs):
return
item = _Item(tuple(text_digest(value) for value in outputs), facts)
self._store(node[1], attempt.item_id, index, item)
except Exception as error:
logger.debug(f"Could not record a cached-run receipt: {error}")
def _store(self, run_key: str, item_id: int, index: int, item: _Item) -> None:
with self._lock:
receipts = self._receipts.pop(run_key, [])
if receipts and receipts[0].item_id == item_id:
receipt = receipts[0]
else:
# The previous job's receipt is complete (its task_done may
# never have fired): it replaces older receipts it equals.
receipts = _deduplicated(receipts)
receipt = _Receipt(item_id)
receipts.insert(0, receipt)
receipt.items[index] = item
self._receipts[run_key] = receipts[: self._max_receipts]
while len(self._receipts) > self._max_keys:
self._receipts.popitem(last=False)
# -- cached nodes -------------------------------------------------------
def replay(
self,
executor: Any,
prompt_id: Any,
nodes: Any,
db_factory: Callable[[], Any],
reader: Optional[Callable[[Any, str, str], Any]] = None,
) -> None:
"""Count the job's cached nodes whose cached text a receipt proves."""
if reader is None:
reader = _host().cached_string_outputs
with self._lock:
attempt = self._attempt
if attempt is None or prompt_id != attempt.prompt_id:
return
if type(nodes) not in (list, tuple) or len(nodes) > MAX_CACHED_NODES:
return
db = None
for node_id in nodes:
node = attempt.nodes.get(node_id) if type(node_id) is str else None
if node is None:
continue
node_type, run_key = node
items = self._match(run_key, reader(executor, node_id, node_type))
if items is None:
logger.debug(
f"Cached node {node_id} not counted: no matching execution"
)
continue
with self._lock:
if node_id in attempt.replayed:
continue # a repeated message or id never counts twice
attempt.replayed.add(node_id)
counts: List[Count] = []
role = graph_role(attempt.graph, node_id, node_type)
for item in items:
seen = graph_seen(attempt.graph, node_id, node_type, item.facts)
counts.extend(own_counts(node_type, item.facts, role, seen))
if not counts:
continue
try:
db = db if db is not None else db_factory()
for count in counts:
if not write_count(db, count):
logger.info(
f"Cached node {node_id}: its {count[0]} no longer exists, "
"not counted"
)
except Exception as error:
logger.warning(f"Could not count cached node {node_id}: {error}")
def _match(self, run_key: str, values: Any) -> Optional[List[_Item]]:
"""The items of the one receipt whose outputs equal ``values``, in order."""
if not values:
return None
length = len(values[0])
if length == 0 or length > self._max_items:
return None
if any(len(slot) != length for slot in values):
return None
wanted = {
index: tuple(text_digest(slot[index]) for slot in values)
for index in range(length)
}
with self._lock:
receipts = self._receipts.get(run_key)
if not receipts:
return None
matches = [receipt for receipt in receipts if receipt.outputs() == wanted]
if not matches:
return None
self._receipts.move_to_end(run_key)
facts = {
tuple(receipt.items[index].facts for index in range(length))
for receipt in matches
}
if len(facts) != 1:
return None
return [matches[0].items[index] for index in range(length)]
def _deduplicated(receipts: List[_Receipt]) -> List[_Receipt]:
"""The newest receipt replaces older ones with the same outputs."""
if not receipts:
return receipts
outputs = receipts[0].outputs()
return [receipts[0]] + [r for r in receipts[1:] if r.outputs() != outputs]
def _job_nodes(graph: Dict[str, Any], outputs: Any) -> Dict[str, Tuple[str, str]]:
"""PromptManager-family nodes the job's outputs need, with their run keys."""
starts = (
[str(node) for node in outputs] if isinstance(outputs, (list, tuple)) else []
)
needed = upstream_nodes(graph, starts)
nodes = {}
for node_id in needed:
node = graph.get(node_id)
node_type = node.get("class_type") if isinstance(node, dict) else None
if node_type not in PROMPT_MANAGER_TYPES:
continue
run_key = prompt_node_key(graph, node_id)
if run_key is not None:
nodes[node_id] = (node_type, run_key)
return nodes
def _host() -> Any:
"""The host adapter module (imported late: it needs ComfyUI's modules)."""
try:
from . import execution_context_host
except ImportError:
from utils import execution_context_host
return execution_context_host
# Shared between the nodes and the host observers
CACHED_RUNS = CachedRuns()
_registered = threading.Event()
_register_lock = threading.Lock()
def register_cached_runs(
server: Any, db_factory: Callable[[], Any], runs: Optional[CachedRuns] = None
) -> bool:
"""Observe dequeue, ``execution_cached`` and ``task_done``; once per process.
Registers nothing when the host lacks any of them: counting is then as
before #347.
"""
runs = CACHED_RUNS if runs is None else runs
with _register_lock:
if _registered.is_set():
return False
queue = getattr(server, "prompt_queue", None)
if not callable(getattr(queue, "get", None)) or not callable(
getattr(queue, "task_done", None)
):
logger.info("Cached runs not counted: the prompt queue is not observable")
return False
try:
from execution import PromptExecutor
except Exception:
logger.info("Cached runs not counted: no ComfyUI executor")
return False
if not callable(getattr(PromptExecutor, "add_message", None)):
logger.info("Cached runs not counted: executor messages unavailable")
return False
subscribe_executor_messages = _host().subscribe_executor_messages
def on_message(executor: Any, args: tuple, kwargs: dict) -> None:
try:
event = args[0] if args else kwargs.get("event")
if event != "execution_cached":
return
server_of = getattr(executor, "server", None)
if getattr(server_of, "prompt_queue", None) is not queue:
return
data = args[1] if len(args) > 1 else kwargs.get("data")
if type(data) is not dict:
return
runs.replay(
executor, data.get("prompt_id"), data.get("nodes"), db_factory
)
except Exception as error:
logger.warning(f"Could not count cached nodes: {error}")
unsubscribers = []
try:
unsubscribers.append(
subscribe_dequeue(queue, "cached_runs", runs.begin_attempt)
)
unsubscribers.append(
subscribe_task_done(
queue,
"cached_runs",
lambda args, kwargs: runs.end_attempt(
args[0] if args else kwargs.get("item_id")
),
)
)
unsubscribers.append(subscribe_executor_messages("cached_runs", on_message))
except Exception:
for unsubscribe in unsubscribers:
unsubscribe()
raise
_registered.set()
logger.info("Counting cached runs of runtime-linked prompt text")
return True
+80 -18
View File
@@ -75,6 +75,82 @@ def subscribe_executor_messages(key: str, callback: Callable) -> Callable[[], No
return state.subscribe(key, callback)
# STRING output slots of each PromptManager-family node, in output order
STRING_SLOTS = {
"PromptManager": (1,),
"PromptManagerFocused": (1,),
"PromptManagerSplit": (1,),
"PromptManagerText": (0,),
"PromptManagerPair": (2, 3),
}
MAX_CACHED_VALUES = 64
def _local_cache_reader(executor: object) -> Callable | None:
"""``node_id -> outputs`` of the executor's local output cache, or None.
Only the supported cache classes, only ``get_local`` (never a provider or
an async lookup), only a real ``CacheEntry`` with at most four outputs.
"""
from comfy_execution import caching
from execution import CacheEntry
cache = executor.caches.outputs
supported = tuple(
cls
for name in ("HierarchicalCache", "LRUCache", "RAMPressureCache")
if isinstance(cls := getattr(caching, name, None), type)
)
if type(cache) not in supported:
return None
local = getattr(cache, "get_local", None)
if not callable(local):
return None
def read(node_id: str) -> list | tuple | None:
entry = local(node_id)
if type(entry) is not CacheEntry:
return None
outputs = entry.outputs
if type(outputs) not in (list, tuple) or not 0 < len(outputs) <= 4:
return None
return outputs
return read
def cached_string_outputs(
executor: object, node_id: str, node_type: str
) -> tuple[tuple[str, ...], ...] | None:
"""One tuple per STRING slot of a cached node's local entry, or None (#347).
None when the cache or entry is not supported, a slot is missing or holds
more than 64 values, any value is not a ``str``, or the slots differ in
length. Never runs a provider or an async lookup.
"""
try:
slots = STRING_SLOTS.get(node_type)
if slots is None or type(node_id) is not str:
return None
local = _local_cache_reader(executor)
outputs = local(node_id) if local is not None else None
if outputs is None or not max(slots) < len(outputs):
return None
found = []
for slot in slots:
values = outputs[slot]
if type(values) not in (list, tuple) or len(values) > MAX_CACHED_VALUES:
return None
if any(type(value) is not str for value in values):
return None
found.append(tuple(values))
if len({len(values) for values in found}) != 1:
return None
return tuple(found)
except Exception:
return None
def observe_cached_context(
executor: object, message_type: object, message_data: object
) -> None:
@@ -101,19 +177,8 @@ def observe_cached_context(
nodes = message_data.get("nodes")
if type(nodes) not in (list, tuple) or len(nodes) > 500:
return
from comfy_execution import caching
from execution import CacheEntry
cache = executor.caches.outputs
supported = tuple(
cls
for name in ("HierarchicalCache", "LRUCache", "RAMPressureCache")
if isinstance(cls := getattr(caching, name, None), type)
)
if type(cache) not in supported:
return
local = getattr(cache, "get_local", None)
if not callable(local):
local = _local_cache_reader(executor)
if local is None:
return
# Only source IDs retained by the bounded submitted graph are queried.
for source in cap.sources:
@@ -124,11 +189,8 @@ def observe_cached_context(
if source.node_type == "PromptManagerPair"
else 0 if source.node_type == "PromptManagerText" else 1
)
entry = local(source.node_id)
if type(entry) is not CacheEntry:
continue
outputs = entry.outputs
if type(outputs) not in (list, tuple) or not slot < len(outputs) <= 4:
outputs = local(source.node_id)
if outputs is None or not slot < len(outputs):
continue
values = outputs[slot]
if type(values) not in (list, tuple) or len(values) > 64:
+56 -8
View File
@@ -300,6 +300,49 @@ def _blank(value: str | bytes | None) -> bool:
return value is None or not value.strip()
def inherited_category(
positive_category: str | bytes | None,
recipe_category: str | bytes | None,
has_recipe: bool,
) -> tuple:
"""``(value, scope)`` of the category an image's link supplies (#332 MC4).
The one rule for the inspector chip, the Images category filter and
History (D39): a linked recipe's own category when it is not blank,
``(recipe category, "pair")``; otherwise the positive's value unchanged,
blank or not, ``(positive category, "positive")``.
"""
if has_recipe and not _blank(recipe_category):
return recipe_category, "pair"
return positive_category, "positive"
class _OwnCategoryUnread:
"""The own-category read failed: hide the category (#332 DC5)."""
def __repr__(self) -> str:
return "OWN_CATEGORY_UNREAD"
OWN_CATEGORY_UNREAD = _OwnCategoryUnread()
def apply_own_category(library: dict, own) -> dict:
"""*library* with an image's own category applied (#332 §5).
``own`` is the own category (a string, shown with scope ``"image"``
whatever the link status), None (the inherited values stay) or
OWN_CATEGORY_UNREAD (category and scope hidden rather than possibly
wrong). Returns a new dict.
"""
result = dict(library)
if own is OWN_CATEGORY_UNREAD:
result.update(category=None, category_scope=None)
elif isinstance(own, str):
result.update(category=own, category_scope="image")
return result
def library_metadata(
positive_category: str | bytes | None,
positive_tags: list,
@@ -312,13 +355,14 @@ def library_metadata(
the linked positive prompt's, the values Set category writes and the Images
filters match. Tag sets are never merged. ``recipe`` is (category, tags).
"""
category, category_scope = positive_category, "positive"
category, category_scope = inherited_category(
positive_category,
recipe[0] if recipe is not None else None,
recipe is not None,
)
tags, tags_scope = positive_tags, "positive"
if recipe is not None:
if not _blank(recipe[0]):
category, category_scope = recipe[0], "pair"
if recipe[1]:
tags, tags_scope = recipe[1], "pair"
if recipe is not None and recipe[1]:
tags, tags_scope = recipe[1], "pair"
return {
"category": category,
"category_scope": category_scope,
@@ -345,9 +389,13 @@ def _library_projection(library: dict) -> dict:
value = library.get(key)
if value is None or isinstance(value, str):
result[key] = value
for key in ("category_scope", "tags_scope"):
# An image's own category (#332) has the scope "image"; tags never do.
for key, scopes in (
("category_scope", ("pair", "positive", "image", None)),
("tags_scope", ("pair", "positive", None)),
):
scope = library.get(key)
if scope in ("pair", "positive", None):
if scope in scopes:
result[key] = scope
tags = library.get("tags")
result["tags"] = (
+50 -14
View File
@@ -51,6 +51,53 @@ def _links(inputs: Any) -> List[str]:
return [value[0] for value in inputs.values() if is_link(value)]
def upstream_closure(
graph: Dict[str, Any], starts: List[str], limit: Optional[int] = None
) -> Optional[Dict[str, Any]]:
"""``{node_id: [class_type, inputs]}`` of ``starts`` and everything upstream.
None when a node or an upstream link's node is missing, or the closure
exceeds ``limit`` nodes.
"""
closure: Dict[str, Any] = {}
stack = list(starts)
while stack:
current = stack.pop()
if current in closure:
continue
node = graph.get(current)
if not isinstance(node, dict):
return None
if limit is not None and len(closure) >= limit:
return None
inputs = node.get("inputs")
closure[current] = [node.get("class_type"), inputs]
stack.extend(_links(inputs))
return closure
def upstream_nodes(graph: Any, starts: List[str]) -> List[str]:
"""Ids of ``starts`` and every node upstream of them that exists in ``graph``.
Used to tell which nodes a job's outputs to execute need ("run selected
output" jobs leave the rest of the graph out). Missing nodes are skipped.
"""
if not isinstance(graph, dict):
return []
found: Dict[str, None] = {}
stack = [str(node) for node in starts]
while stack:
current = stack.pop()
if current in found:
continue
node = graph.get(current)
if not isinstance(node, dict):
continue
found[current] = None
stack.extend(_links(node.get("inputs")))
return list(found)
def prompt_node_key(graph: Any, node_id: Any) -> Optional[str]:
"""Digest of ``node_id`` and its whole upstream closure, or None.
@@ -62,20 +109,9 @@ def prompt_node_key(graph: Any, node_id: Any) -> Optional[str]:
if not isinstance(graph, dict) or node_id is None:
return None
start = str(node_id)
closure: Dict[str, Any] = {}
stack = [start]
while stack:
current = stack.pop()
if current in closure:
continue
node = graph.get(current)
if not isinstance(node, dict):
return None
if len(closure) >= MAX_VISITS:
return None
inputs = node.get("inputs")
closure[current] = [node.get("class_type"), inputs]
stack.extend(_links(inputs))
closure = upstream_closure(graph, [start], MAX_VISITS)
if closure is None:
return None
try:
encoded = json.dumps(
[RUN_KEY_VERSION, start, closure],
+18 -7
View File
@@ -16,7 +16,9 @@ it never counts. Servers without a ``prompt_queue`` fall back to counting at
submit time.
Text from known pure string nodes is resolved at queue time too. Anything else
(PromptSearchList batches, unknown nodes) is counted by the node as it executes.
(PromptSearchList batches, unknown nodes) is counted by the node as it executes,
and when ComfyUI serves such a node from its cache, the job replays what that
execution counted (``cached_runs``, #347).
"""
import threading
@@ -167,13 +169,22 @@ def queued_pair_runs(graph: Any) -> List[Tuple[str, str, str]]:
def find_pair_id(db: Any, positive: str, negative: str) -> Optional[int]:
"""Find a saved pair without creating or overwriting its metadata."""
positive_row = db.get_prompt_by_hash(generate_prompt_hash(positive))
negative_row = (
db.get_prompt_by_hash(generate_prompt_hash(negative))
if negative.strip()
else None
return find_pair_id_by_hashes(
db,
generate_prompt_hash(positive),
generate_prompt_hash(negative) if negative.strip() else None,
)
if not positive_row or (negative.strip() and not negative_row):
def find_pair_id_by_hashes(
db: Any, positive_hash: str, negative_hash: Optional[str]
) -> Optional[int]:
"""The saved pair of two prompt hashes; ``None`` negative is the empty one."""
positive_row = db.get_prompt_by_hash(positive_hash)
negative_row = (
db.get_prompt_by_hash(negative_hash) if negative_hash is not None else None
)
if not positive_row or (negative_hash is not None and not negative_row):
return None
row = (
db.model.get_connection()
+6 -1
View File
@@ -20,6 +20,8 @@ const REPRESENTATIONS = {
limited: 'Record exceeds inspection limits; text is unavailable',
};
const TEXTLESS = new Set(['invalid', 'limited']);
// #332: the hidden prefix that names where a category chip's value came from.
const CATEGORY_PREFIX = {image: 'Image category: ', positive: 'Positive prompt category: ', pair: 'Recipe category: '};
const REVEAL_URL = '/prompt_manager/images/reveal';
// D38 Show in folder: the route's answers, in the maintainer's words.
const REVEAL_OPENED = 'Opened the folder on this computer.';
@@ -265,7 +267,10 @@ export function mountFileInspector(container, {api, sender, parentScope, variant
inner.append(heading, el('p', filename, 'small'));
const badges = el('div', undefined, 'tags'), badge = evidenceBadge(record.prompts);
if (badge) badges.append(el('span', badge, 'chip blue'));
if (linked && library.category) badges.append(prefixedChip(library.category_scope === 'pair' ? 'Recipe category: ' : 'Positive prompt category: ', library.category));
// #332: an image's own category shows for any link status; an inherited one only when linked.
if (library?.category && (library.category_scope === 'image' || linked)) {
badges.append(prefixedChip(CATEGORY_PREFIX[library.category_scope] ?? CATEGORY_PREFIX.positive, library.category));
}
if (badges.childNodes.length) inner.append(badges);
const prompts = el('div'); inner.append(prompts);
const blocks = mountPromptBlocks(prompts, {sender, parentScope:contentScope, includePairAction:false, contextLabel:promptContext, title});
+3 -1
View File
@@ -201,7 +201,9 @@ function resolutionFor(value) {
function libraryFor(value) {
if (!object(value) || !['linked', 'none', 'ambiguous', 'unavailable', 'limited'].includes(value.status) || value.scope !== 'saved-library' ||
!Array.isArray(value.tags) || value.tags.some(tag => typeof tag !== 'string') ||
![value.category_scope, value.tags_scope].every(scope => [null, 'pair', 'positive'].includes(scope))) throw new TypeError('Invalid inspection library');
![null, 'pair', 'positive', 'image'].includes(value.category_scope) ||
![null, 'pair', 'positive'].includes(value.tags_scope)) throw new TypeError('Invalid inspection library');
// #332: an image's own category (scope "image") may sit on any link status; tags never have that scope.
// D39: each field names its own source, so a recipe's category can sit beside its positive's tags.
const library = { status: value.status, scope: value.scope, tags: [...value.tags], category_scope: value.category_scope, tags_scope: value.tags_scope };
for (const key of ['positive_id', 'pair_id', 'title', 'category']) {