fix(browser): server shutdown and its tests on Windows, Python < 3.13 and CI

Three CI failures on ed809f7:

- Windows: server.py exited 1 on stdin EOF because create_app works with
  the temp root as its cwd, and Windows cannot remove a process's cwd.
  main() now restores the original cwd before the root is removed.
- Python 3.10 to 3.12: the lifetime test's cleanup called communicate()
  after closing stdin, which flushes the closed pipe (ValueError); 3.13
  tolerates it. It now waits and closes the pipes itself, and the exit
  assertion shows the server's stderr. Under 3.12: errors=2 before, OK
  after.
- Tooling (CI container): the SIGKILLed-worker tests treated an exited
  server as alive, because the container's PID 1 does not reap orphans
  and kill(pid, 0) succeeds on a zombie. Reproduced locally with docker
  without --init (2 failed); a zombie now counts as exited (2 passed).

.coveragerc

@@ -14,6 +14,10 @@ omit =
     */node_modules/*
     */static/*
     */media/*
+    */htmlcov/*
+    */.claude/*
+    restart_gallery.py
+    standalone_tagger.py
     
 branch = true
 parallel = true

.flake8

@@ -0,0 +1,6 @@
+[flake8]
+max-line-length = 88
+extend-ignore = E203, W503
+exclude = reference/, .git, __pycache__, web/, venv/, htmlcov/, build/, dist/, .claude/, docs/, standalone_tagger.py
+per-file-ignores =
+    tests/*:E402

.github/branch-protection.json

@@ -0,0 +1,30 @@
+{
+  "required_status_checks": {
+    "strict": true,
+    "contexts": [
+      "py3.10 on ubuntu-latest",
+      "py3.11 on ubuntu-latest",
+      "py3.12 on ubuntu-latest",
+      "py3.13 on ubuntu-latest",
+      "py3.10 on windows-latest",
+      "py3.11 on windows-latest",
+      "py3.12 on windows-latest",
+      "py3.13 on windows-latest",
+      "py3.10 on macos-latest",
+      "py3.11 on macos-latest",
+      "py3.12 on macos-latest",
+      "py3.13 on macos-latest",
+      "coverage gate (85%)",
+      "black and flake8",
+      "frontend modules",
+      "viewerjs",
+      "real ComfyUI compatibility",
+      "bandit (high severity)"
+    ]
+  },
+  "enforce_admins": true,
+  "required_pull_request_reviews": null,
+  "restrictions": null,
+  "allow_force_pushes": false,
+  "allow_deletions": false
+}

.github/dependabot.yml

@@ -46,3 +46,15 @@ updates:
         update-types:
           - "minor"
           - "patch"
+
+  # Maintainer browser test dependencies; Viewer.js itself has an asset-aware updater.
+  - package-ecosystem: "npm"
+    directory: "/tools/vendor"
+    schedule:
+      interval: "weekly"
+    allow:
+      - dependency-name: "@playwright/test"
+    labels:
+      - "dependencies"
+    commit-message:
+      prefix: "deps(vendor-tools)"

.github/workflows/code-quality.yml

@@ -9,42 +9,15 @@ on:
   pull_request:
     branches: [main]
 
+# Superseded PR pushes cancel their older run. Every main push gets its own
+# group (run_id), so no main commit's run is ever cancelled or dropped as pending.
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event_name == 'pull_request' && github.ref || github.run_id }}
+  cancel-in-progress: ${{ github.event_name == 'pull_request' }}
+
 jobs:
-  lint:
-    runs-on: ubuntu-latest
-
-    steps:
-      - uses: actions/checkout@v7
-
-      - name: Set up Python 3.10
-        uses: actions/setup-python@v7
-        with:
-          python-version: "3.10"
-
-      - name: Cache pip dependencies
-        uses: actions/cache@v6
-        with:
-          path: ~/.cache/pip
-          key: ${{ runner.os }}-pip-quality-${{ hashFiles('requirements-dev.txt') }}
-          restore-keys: |
-            ${{ runner.os }}-pip-quality-
-
-      - name: Install dependencies
-        run: |
-          python -m pip install --upgrade pip
-          pip install -r requirements-dev.txt
-
-      - name: Check formatting with Black
-        run: black --check --diff .
-
-      - name: Lint with flake8
-        run: |
-          # Fail on syntax errors and undefined names
-          flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics
-          # Warnings as non-blocking
-          flake8 . --count --exit-zero --max-line-length=88 --statistics --exclude=reference/,.git,__pycache__,web/
-
   security:
+    name: bandit (high severity)
     runs-on: ubuntu-latest
 
     steps:
@@ -58,5 +31,9 @@ jobs:
       - name: Install bandit
         run: pip install bandit[toml]
 
-      - name: Security scan
-        run: bandit -r . -ll --exclude=./tests,./reference,./venv -q || true
+      # Medium findings are reported without failing until they are triaged.
+      # High severity is a hard gate.
+      - name: Security report (medium and above, informational)
+        run: bandit -r . -ll -q --exclude ./tests,./reference,./venv,./tools/vendor/node_modules --exit-zero
+      - name: Security scan (high severity gate)
+        run: bandit -r . -lll -q --exclude ./tests,./reference,./venv,./tools/vendor/node_modules

.github/workflows/test.yml

@@ -8,13 +8,26 @@ on:
     branches: [main]
   push:
     branches: [main]
+  workflow_dispatch:
+
+# Superseded PR pushes cancel their older run. Every main push gets its own
+# group (run_id), so no main commit's run is ever cancelled or dropped as pending.
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event_name == 'pull_request' && github.ref || github.run_id }}
+  cancel-in-progress: ${{ github.event_name == 'pull_request' }}
+
+env:
+  PYTHONFAULTHANDLER: "1"
 
 jobs:
   test:
-    runs-on: ubuntu-latest
+    name: py${{ matrix.python-version }} on ${{ matrix.os }}
+    runs-on: ${{ matrix.os }}
     strategy:
+      fail-fast: false
       matrix:
-        python-version: ["3.10", "3.11", "3.12"]
+        os: [ubuntu-latest, windows-latest, macos-latest]
+        python-version: ["3.10", "3.11", "3.12", "3.13"]
 
     steps:
       - uses: actions/checkout@v7
@@ -23,14 +36,8 @@ jobs:
         uses: actions/setup-python@v7
         with:
           python-version: ${{ matrix.python-version }}
-
-      - name: Cache pip dependencies
-        uses: actions/cache@v6
-        with:
-          path: ~/.cache/pip
-          key: ${{ runner.os }}-pip-${{ matrix.python-version }}-${{ hashFiles('requirements-dev.txt') }}
-          restore-keys: |
-            ${{ runner.os }}-pip-${{ matrix.python-version }}-
+          cache: pip
+          cache-dependency-path: requirements-dev.txt
 
       - name: Install dependencies
         run: |
@@ -38,13 +45,165 @@ jobs:
           pip install -r requirements-dev.txt
 
       - name: Run tests
-        run: python -m unittest discover tests/ -v
+        run: python -m unittest discover -s tests -v
 
-  js-test:
+  comfyui:
+    name: real ComfyUI compatibility
+    runs-on: ubuntu-latest
+    timeout-minutes: 25
+    steps:
+      - uses: actions/checkout@v7
+        with:
+          persist-credentials: false
+      - uses: actions/setup-node@v7
+        with:
+          node-version: "24"
+          cache: npm
+          cache-dependency-path: tools/vendor/package-lock.json
+      - name: Read the reviewed runtime pins
+        id: runtime
+        run: |
+          node --input-type=module <<'JS' >> "$GITHUB_OUTPUT"
+          import runtime from './tools/vendor/comfyui-runtime.json' with { type: 'json' };
+          for (const [key, value] of Object.entries(runtime)) console.log(`${key}=${value}`);
+          import crossPack from './tools/vendor/cross-pack-runtime.json' with { type: 'json' };
+          console.log(`kikoRepository=${crossPack.repository}`);
+          console.log(`kikoRevision=${crossPack.revision}`);
+          JS
+      - name: Check out pinned ComfyUI
+        uses: actions/checkout@v7
+        with:
+          repository: ${{ steps.runtime.outputs.repository }}
+          ref: ${{ steps.runtime.outputs.revision }}
+          path: .ci-comfyui
+          persist-credentials: false
+      - name: Check out pinned KikoTools for compatibility
+        uses: actions/checkout@v7
+        with:
+          repository: ${{ steps.runtime.outputs.kikoRepository }}
+          ref: ${{ steps.runtime.outputs.kikoRevision }}
+          path: .ci-kikotools
+          persist-credentials: false
+      - uses: actions/setup-python@v7
+        with:
+          python-version: ${{ steps.runtime.outputs.pythonVersion }}
+          cache: pip
+          cache-dependency-path: |
+            requirements-dev.txt
+            .ci-comfyui/requirements.txt
+            tools/vendor/comfyui-runtime.json
+      - name: Install CPU runtime and test dependencies
+        run: |
+          python -m pip install --upgrade pip
+          pip install "torch==${{ steps.runtime.outputs.torchVersion }}" "torchvision==${{ steps.runtime.outputs.torchvisionVersion }}" --index-url https://download.pytorch.org/whl/cpu
+          pip install -r .ci-comfyui/requirements.txt -r requirements-dev.txt "torch==${{ steps.runtime.outputs.torchVersion }}" "torchvision==${{ steps.runtime.outputs.torchvisionVersion }}"
+      - run: npm ci --prefix tools/vendor --ignore-scripts
+      - run: npx --prefix tools/vendor playwright install --with-deps chromium
+      - name: Classic and Nodes 2.0 startup and compatibility
+        env:
+          PM_COMFY_ROOT: ${{ github.workspace }}/.ci-comfyui
+          PM_TEST_PYTHON: python
+        run: npm run --prefix tools/vendor test:comfyui
+      - name: Real cached and headless execution history
+        env:
+          PM_COMFY_ROOT: ${{ github.workspace }}/.ci-comfyui
+          PM_TEST_PYTHON: python
+        run: npm run --prefix tools/vendor test:executions
+      - name: Independent and combined pack compatibility
+        env:
+          PM_COMFY_ROOT: ${{ github.workspace }}/.ci-comfyui
+          PM_KIKO_ROOT: ${{ github.workspace }}/.ci-kikotools
+          PM_TEST_PYTHON: python
+        run: npm run --prefix tools/vendor test:compat
+      - name: Upload real-host reports and startup logs
+        if: always()
+        uses: actions/upload-artifact@v7
+        with:
+          name: real-comfyui-reports
+          path: tools/vendor/test-results/
+          if-no-files-found: ignore
+          retention-days: 14
+
+  coverage:
+    name: coverage gate (85%)
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v7
 
-      # Pure-JS frontend modules, tested with the runner's preinstalled Node
+      - uses: actions/setup-python@v7
+        with:
+          python-version: "3.12"
+          cache: pip
+          cache-dependency-path: requirements-dev.txt
+
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+          pip install -r requirements-dev.txt
+
+      - name: Measure coverage
+        run: |
+          python -m coverage run -m unittest discover -s tests
+          python -m coverage combine
+          python -m coverage report --fail-under=85
+
+  perf-budget:
+    # The reference wall-clock measurement (TESTING.md "Performance budgets").
+    # Shared-runner timing varies, so it never gates a pull request: it runs on
+    # each push to main and on demand. A red run on main means fix the code.
+    name: perf budget
+    if: github.event_name == 'push' || github.event_name == 'workflow_dispatch'
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v7
+
+      - uses: actions/setup-python@v7
+        with:
+          python-version: "3.13"
+          cache: pip
+          cache-dependency-path: requirements-dev.txt
+
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+          pip install -r requirements-dev.txt
+
+      - name: Workspace filter budget
+        env:
+          PROMPTMANAGER_PERF_BUDGET: "1"
+        run: python -m unittest -v tests.test_api_workspace_images.TestWorkspaceImages.test_filter_budget
+
+  lint:
+    name: black and flake8
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v7
+      - uses: actions/setup-python@v7
+        with:
+          python-version: "3.12"
+      - run: pip install black flake8
+      - run: black --check .
+      - run: flake8 .
+
+  js-test:
+    name: frontend modules
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v7
+
+      - uses: actions/setup-node@v7
+        with:
+          node-version: "24"
+
+      - name: Check shipped extension and page modules parse
+        run: |
+          shopt -s globstar nullglob
+          for f in web/comfy/**/*.js web/comfy/**/*.mjs web/js/**/*.js web/js/**/*.mjs; do
+            node --input-type=module --check < "$f" || exit 1
+          done
+
+      - name: Install pinned shared test contracts
+        run: npm ci --prefix tools/vendor --ignore-scripts
+
       - name: Run JS tests
         run: node --test tests/js/*.test.js

.github/workflows/viewerjs-check.yml

@@ -0,0 +1,238 @@
+name: Viewer.js vendor and compatibility
+
+on:
+  pull_request:
+    branches: [main]
+  push:
+    branches: [main]
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+# Superseded PR pushes cancel their older run. Every main push gets its own
+# group (run_id), so no main commit's run is ever cancelled or dropped as pending.
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event_name == 'pull_request' && github.ref || github.run_id }}
+  cancel-in-progress: ${{ github.event_name == 'pull_request' }}
+
+# Timeouts are the measured job time plus at least 50%, and never under 20 minutes for a browser job,
+# because runner speed varies. Before sharding (main 2cecfcf, one worker): frontend coverage 32m10s,
+# page workflows firefox 18m59s, webkit 19m26s. First sharded run (PR #274 run 39712909047, 3 workers,
+# no npm cache, where npm ci stalled for 5m in six of eight jobs): coverage shards 11m19s/10m47s/10m49s
+# (tests 5m20s/4m42s/4m46s), merge 5m52s (report and check 4s), firefox shards 7m53s/9m27s (tests
+# 4m41s/3m29s), webkit shards 12m25s+ (cancelled at the old 12-minute limit after its 267 tests passed in
+# 6m29s) and 4m53s (tests 3m08s). With the npm cache hit (run 39725176320 at cbc8a61, npm ci 0-2s):
+# 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.
+jobs:
+  frontend-coverage:
+    name: frontend coverage (shard ${{ matrix.shard }}/3)
+    runs-on: ubuntu-latest
+    # Layout comparisons depend on the fonts each browser falls back to, so browser jobs run in the
+    # same pinned Playwright image as local runs (tag = @playwright/test in tools/vendor/package-lock.json;
+    # tests/vendor/browser-image.test.mjs checks it). Browsers and fonts come from the image.
+    container:
+      image: mcr.microsoft.com/playwright:v1.64.1-noble
+      options: --user 1001
+    timeout-minutes: 20
+    # Public-repo ubuntu-latest runners have 4 vCPUs: three Playwright workers, each with its own test
+    # server, leave one CPU for the servers and the runner.
+    env:
+      PM_TEST_WORKERS: 3
+    strategy:
+      fail-fast: false
+      matrix:
+        shard: [1, 2, 3]
+    steps:
+      - uses: actions/checkout@v7
+      # The image's own Python 3.12 and Node 24: setup-python/setup-node install the runner's toolcache
+      # builds, whose entry points name host paths that do not exist in the container. The image has no
+      # ensurepip and the job has no root, so a checksum-pinned pip wheel bootstraps a user-space venv.
+      - name: Python environment
+        env:
+          PIP_WHEEL: https://files.pythonhosted.org/packages/f3/6e/1736e5b4ae2b778ef2f81c47d797de9f891d4d8acb047a24ca37a60294dd/pip-26.2.1-py3-none-any.whl
+          PIP_SHA256: 71138adf1f4ca900cdb7d289c21b7494329f2332b6d85f0e1c42108c0384ed3e
+        run: |
+          venv="$RUNNER_TEMP/venv" wheel="$RUNNER_TEMP/pip-26.2.1-py3-none-any.whl"
+          python3 -m venv --without-pip "$venv"
+          curl -fsSL --retry 3 -o "$wheel" "$PIP_WHEEL"
+          echo "$PIP_SHA256  $wheel" | sha256sum -c -
+          "$venv/bin/python" "$wheel/pip" install --no-index --no-cache-dir "$wheel"
+          "$venv/bin/python" -m pip install --no-cache-dir -r requirements-dev.txt
+          echo "$venv/bin" >> "$GITHUB_PATH"
+      - run: python --version && node --version && node --version | grep -q '^v24\.'
+      # npm's download cache, keyed on the lockfile; inside the container ~ is the job's HOME (/github/home),
+      # which npm also uses. node_modules itself is still built fresh by npm ci.
+      - name: Cache npm downloads
+        uses: actions/cache@v6
+        with:
+          path: ~/.npm
+          key: npm-${{ runner.os }}-playwright-v1.64.1-${{ hashFiles('tools/vendor/package-lock.json') }}
+      - run: npm ci --prefix tools/vendor --ignore-scripts
+      # Shard 1 also runs the node units and the viewer suite, so the vendor pins and the tooling tests
+      # run on the lightest shard instead.
+      - if: matrix.shard == 3
+        run: npm run --prefix tools/vendor check
+      - if: matrix.shard == 3
+        run: npm run --prefix tools/vendor test
+      - name: Page workflows with coverage, shard ${{ matrix.shard }}/3
+        run: npm run --prefix tools/vendor test:coverage:shard -- ${{ matrix.shard }}/3
+      - name: Upload raw coverage
+        uses: actions/upload-artifact@v7
+        with:
+          name: coverage-raw-${{ matrix.shard }}
+          path: coverage/js/raw/
+          if-no-files-found: error
+          retention-days: 3
+      - name: Upload Chromium and touch traces
+        if: always()
+        uses: actions/upload-artifact@v7
+        with:
+          name: frontend-traces-shard-${{ matrix.shard }}
+          path: tools/vendor/test-results/
+          retention-days: 14
+
+  frontend-coverage-merge:
+    name: frontend coverage (merge)
+    needs: frontend-coverage
+    runs-on: ubuntu-latest
+    # Same image as the shards: the raw files name sources by the container's workspace path.
+    container:
+      image: mcr.microsoft.com/playwright:v1.64.1-noble
+      options: --user 1001
+    timeout-minutes: 10
+    steps:
+      - uses: actions/checkout@v7
+      - run: node --version && node --version | grep -q '^v24\.'
+      # npm's download cache, keyed on the lockfile; inside the container ~ is the job's HOME (/github/home),
+      # which npm also uses. node_modules itself is still built fresh by npm ci.
+      - name: Cache npm downloads
+        uses: actions/cache@v6
+        with:
+          path: ~/.npm
+          key: npm-${{ runner.os }}-playwright-v1.64.1-${{ hashFiles('tools/vendor/package-lock.json') }}
+      - run: npm ci --prefix tools/vendor --ignore-scripts
+      # Each artifact lands in its own coverage-raw-N directory, so the next step can count them.
+      - uses: actions/download-artifact@v8
+        with:
+          pattern: coverage-raw-*
+          path: ${{ runner.temp }}/coverage-raw
+      # coverage:merge only checks that raw files exist; merging two of three shards would report
+      # partial coverage, so a missing or empty shard artifact fails here.
+      - name: Verify every shard's raw coverage arrived
+        shell: bash
+        env:
+          STAGED_RAW: ${{ runner.temp }}/coverage-raw
+          EXPECTED_SHARDS: 3
+        run: |
+          found=0
+          for shard in $(seq 1 "$EXPECTED_SHARDS"); do
+            directory="$STAGED_RAW/coverage-raw-$shard"
+            count=0
+            if [ -d "$directory" ]; then count=$(find "$directory" -maxdepth 1 -type f -name '*.json' | wc -l); fi
+            echo "coverage-raw-$shard: $count raw files"
+            if [ "$count" -gt 0 ]; then found=$((found + 1)); fi
+          done
+          echo "shard artifacts with raw coverage: $found of $EXPECTED_SHARDS"
+          if [ "$found" -ne "$EXPECTED_SHARDS" ]; then
+            echo "::error::Expected raw coverage from $EXPECTED_SHARDS shards, got $found of $EXPECTED_SHARDS"
+            exit 1
+          fi
+          staged=$(find "$STAGED_RAW" -mindepth 2 -maxdepth 2 -type f -name '*.json' | wc -l)
+          mkdir -p coverage/js/raw
+          # cp -n never overwrites; a clash shows up as fewer merged files than staged ones.
+          find "$STAGED_RAW" -mindepth 2 -maxdepth 2 -type f -name '*.json' -exec cp -n -t coverage/js/raw {} + || true
+          merged=$(find coverage/js/raw -maxdepth 1 -type f -name '*.json' | wc -l)
+          if [ "$merged" -ne "$staged" ]; then
+            echo "::error::Shards wrote clashing raw file names: $staged staged, $merged merged"
+            exit 1
+          fi
+          echo "merged $merged raw files"
+      - name: Frontend coverage report and baseline check
+        run: npm run --prefix tools/vendor coverage:merge
+      - name: Upload frontend coverage report
+        if: always()
+        uses: actions/upload-artifact@v7
+        with:
+          name: frontend-test-reports
+          path: |
+            coverage/js/
+            !coverage/js/raw/**
+          retention-days: 14
+
+  browser-workflows:
+    name: page workflows (${{ matrix.browser }}, shard ${{ matrix.shard }}/2)
+    runs-on: ubuntu-latest
+    # Layout comparisons depend on the fonts each browser falls back to, so browser jobs run in the
+    # same pinned Playwright image as local runs (tag = @playwright/test in tools/vendor/package-lock.json;
+    # tests/vendor/browser-image.test.mjs checks it). Browsers and fonts come from the image.
+    container:
+      image: mcr.microsoft.com/playwright:v1.64.1-noble
+      options: --user 1001
+    timeout-minutes: 20
+    # Public-repo ubuntu-latest runners have 4 vCPUs: three Playwright workers, each with its own test
+    # server, leave one CPU for the servers and the runner.
+    env:
+      PM_TEST_WORKERS: 3
+    strategy:
+      fail-fast: false
+      matrix:
+        browser: [firefox, webkit]
+        shard: [1, 2]
+    steps:
+      - uses: actions/checkout@v7
+      # The image's own Python 3.12 and Node 24: setup-python/setup-node install the runner's toolcache
+      # builds, whose entry points name host paths that do not exist in the container. The image has no
+      # ensurepip and the job has no root, so a checksum-pinned pip wheel bootstraps a user-space venv.
+      - name: Python environment
+        env:
+          PIP_WHEEL: https://files.pythonhosted.org/packages/f3/6e/1736e5b4ae2b778ef2f81c47d797de9f891d4d8acb047a24ca37a60294dd/pip-26.2.1-py3-none-any.whl
+          PIP_SHA256: 71138adf1f4ca900cdb7d289c21b7494329f2332b6d85f0e1c42108c0384ed3e
+        run: |
+          venv="$RUNNER_TEMP/venv" wheel="$RUNNER_TEMP/pip-26.2.1-py3-none-any.whl"
+          python3 -m venv --without-pip "$venv"
+          curl -fsSL --retry 3 -o "$wheel" "$PIP_WHEEL"
+          echo "$PIP_SHA256  $wheel" | sha256sum -c -
+          "$venv/bin/python" "$wheel/pip" install --no-index --no-cache-dir "$wheel"
+          "$venv/bin/python" -m pip install --no-cache-dir -r requirements-dev.txt
+          echo "$venv/bin" >> "$GITHUB_PATH"
+      - run: python --version && node --version && node --version | grep -q '^v24\.'
+      # npm's download cache, keyed on the lockfile; inside the container ~ is the job's HOME (/github/home),
+      # which npm also uses. node_modules itself is still built fresh by npm ci.
+      - name: Cache npm downloads
+        uses: actions/cache@v6
+        with:
+          path: ~/.npm
+          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
+        run: npm run --prefix tools/vendor test:workflows -- --project ${{ matrix.browser }} --shard ${{ matrix.shard }}/2
+      - name: Upload browser traces
+        if: always()
+        uses: actions/upload-artifact@v7
+        with:
+          name: browser-traces-${{ matrix.browser }}-shard-${{ matrix.shard }}
+          path: tools/vendor/test-results/
+          retention-days: 14
+
+  # Branch protection requires the exact context "viewerjs". This job keeps that
+  # name and fails unless every frontend job above succeeded (every shard, the
+  # coverage merge and every browser shard), including when one was cancelled or skipped.
+  viewerjs:
+    needs: [frontend-coverage, frontend-coverage-merge, browser-workflows]
+    if: always()
+    runs-on: ubuntu-latest
+    timeout-minutes: 5
+    steps:
+      - name: Require every frontend job to succeed
+        env:
+          COVERAGE_RESULT: ${{ needs.frontend-coverage.result }}
+          MERGE_RESULT: ${{ needs.frontend-coverage-merge.result }}
+          WORKFLOWS_RESULT: ${{ needs.browser-workflows.result }}
+        run: |
+          echo "frontend-coverage: $COVERAGE_RESULT"
+          echo "frontend-coverage-merge: $MERGE_RESULT"
+          echo "browser-workflows: $WORKFLOWS_RESULT"
+          test "$COVERAGE_RESULT" = success && test "$MERGE_RESULT" = success && test "$WORKFLOWS_RESULT" = success

.github/workflows/viewerjs-update.yml

@@ -0,0 +1,83 @@
+name: Update vendored Viewer.js
+
+on:
+  schedule:
+    - cron: '30 7 * * *'
+  workflow_dispatch:
+    inputs:
+      version:
+        description: 'Stable version to vendor; leave blank for the latest npm release'
+        required: false
+        type: string
+
+permissions:
+  contents: read
+
+concurrency:
+  group: viewerjs-update
+  cancel-in-progress: false
+
+jobs:
+  update:
+    runs-on: ubuntu-latest
+    permissions:
+      contents: write
+      pull-requests: write
+      actions: write
+    env:
+      GH_TOKEN: ${{ github.token }}
+    steps:
+      - uses: actions/checkout@v7
+        with:
+          ref: ${{ github.event.repository.default_branch }}
+      - uses: actions/setup-node@v7
+        with:
+          node-version: '24'
+          cache: npm
+          cache-dependency-path: tools/vendor/package-lock.json
+      - run: npm ci --prefix tools/vendor --ignore-scripts
+      - name: Update the dependency and regenerate assets
+        env:
+          VIEWER_VERSION: ${{ inputs.version }}
+        run: |
+          if [ -n "$VIEWER_VERSION" ]; then
+            npm run --prefix tools/vendor update -- --version "$VIEWER_VERSION"
+          else
+            npm run --prefix tools/vendor update
+          fi
+      - name: Detect changes
+        id: changes
+        run: |
+          if git diff --quiet -- tools/vendor/package.json tools/vendor/package-lock.json web/lib/viewerjs; then
+            echo 'changed=false' >> "$GITHUB_OUTPUT"
+          else
+            echo 'changed=true' >> "$GITHUB_OUTPUT"
+          fi
+      - if: steps.changes.outputs.changed == 'true'
+        run: |
+          npm run --prefix tools/vendor check
+          npm run --prefix tools/vendor test
+          node --test tests/js/*.test.js
+          npx --prefix tools/vendor playwright install --with-deps chromium
+          npm run --prefix tools/vendor test:browser
+      - name: Open the update PR
+        if: steps.changes.outputs.changed == 'true'
+        env:
+          BASE_BRANCH: ${{ github.event.repository.default_branch }}
+        run: |
+          version=$(node -p "require('./web/lib/viewerjs/provenance.json').version")
+          branch="deps/viewerjs-$version"
+          existing=$(gh pr list --repo "$GITHUB_REPOSITORY" --state all --head "$branch" --json number --jq 'length')
+          if [ "$existing" != '0' ]; then
+            echo "A PR already exists for Viewer.js $version; leaving it for review."
+            exit 0
+          fi
+          remote=$(git ls-remote origin "refs/heads/$branch")
+          if [ -z "$remote" ]; then
+            git switch -c "$branch"
+            git add tools/vendor/package.json tools/vendor/package-lock.json web/lib/viewerjs
+            git -c user.name='github-actions[bot]' -c user.email='41898282+github-actions[bot]@users.noreply.github.com' commit -m "deps: update vendored Viewer.js to $version"
+            git push origin "HEAD:refs/heads/$branch"
+          fi
+          gh pr create --repo "$GITHUB_REPOSITORY" --base "$BASE_BRANCH" --head "$branch" --title "deps: update vendored Viewer.js to $version" --body-file tools/vendor/update-pr.md
+          gh workflow run viewerjs-check.yml --repo "$GITHUB_REPOSITORY" --ref "$branch"

.gitignore

@@ -9,6 +9,12 @@ AGENTS.md
 CLAUDE.md
 
 # Local files
+# Local PromptManager state (config.json holds API keys and machine paths)
+config.json
+resume.sh
+*.backup_*
+promptmanager_logs/
+custom_data/
 data/
 docs/
 logs/
@@ -194,3 +200,14 @@ cython_debug/
 
 # PyPI configuration file
 .pypirc
+
+# Maintainer-only JavaScript tooling and browser test artifacts
+node_modules/
+test-results/
+playwright-report/
+
+# Combined frontend coverage reports
+coverage/
+
+# Disposable upstream checkout for real-host CI
+.ci-comfyui/

.pre-commit-config.yaml

@@ -9,7 +9,7 @@ repos:
 
       - id: flake8
         name: flake8
-        entry: flake8 --select=E9,F63,F7,F82 --show-source --exclude=reference/,web/
+        entry: flake8 --show-source
         language: system
         types: [python]
 

README.md

@@ -139,6 +139,35 @@ The comprehensive gallery system provides complete management of all ComfyUI out
 
 _Powered by [ViewerJS](https://github.com/fengyuanchen/viewerjs) for professional image viewing capabilities._
 
+Viewer.js is vendored locally. See [Viewer.js maintenance](tools/vendor/README.md)
+for reproducible updates, verification, and automated update PRs.
+
+Prompt Manager appears as **Prompt Manager (Classic)** in the Add Node menu.
+It keeps the existing layout and behavior while using shared
+[ComfyAssets Toolkit](https://github.com/ComfyAssets/ComfyAssets-Toolkit) controls
+that follow ComfyUI's light/dark theme. Existing workflows keep the same internal
+node identifier and sockets. The reusable architecture is inspired by
+[Pixaroma](https://github.com/pixaroma/ComfyUI-Pixaroma); no implementation or assets
+were copied. See the [toolkit maintenance notes](tools/vendor/README.md#shared-comfyassets-browser-controls).
+
+**Prompt Manager (Focused)** adds an editor-first layout with collapsible
+composition and metadata fields. Open **Browse library** to search saved prompts,
+filter by saved category and comma-separated tags, preview a selection, then choose
+**Use prompt**; **Load metadata** controls whether
+category and tags come with it. Metadata fields suggest saved categories and tags;
+choose with the mouse or arrow keys and Enter. Tag selections append `, ` for the
+next tag; category remains a single value. New values can still be typed freely.
+Prepend and append STRING sockets remain visible
+when the fields are collapsed, so LoRA trigger outputs can connect directly.
+Encoding, storage, and CONDITIONING/text outputs match Classic. Both layouts
+support ComfyUI Classic and Nodes 2.0.
+
+**Prompt Manager (Split)** keeps the same editor and library visible together.
+Wide nodes place the library beside the editor; narrow nodes stack the panels.
+It shares Focused's category/tag autocomplete, search filters, preview-before-use,
+workflow persistence, and connectable prepend/append inputs. Encoding and storage
+match Classic and Focused.
+
 ### 🔬 ComfyUI Workflow Analysis
 
 - **PNG Metadata Parsing**: Advanced PNG chunk analysis to extract embedded ComfyUI data
@@ -164,7 +193,7 @@ _Powered by [ViewerJS](https://github.com/fengyuanchen/viewerjs) for professiona
 
 - **Full-text search** across all stored prompts
 - **Category filtering** for organized prompt collections
-- **Tag-based search** with support for multiple tags
+- **Tag-based search** with support for multiple tags, with autocomplete that narrows to tags seen together with the ones already typed
 - **Rating filters** to find your best prompts
 - **Date range filtering** for temporal searches
 - **Recent prompts** quick access
@@ -306,9 +335,12 @@ The comprehensive gallery system provides complete access to all your ComfyUI ou
 
 Configure your viewing experience in the settings panel:
 
-- **Performance Settings**: Grid columns, image quality, lazy loading
+- **Display Settings**: Default view, items per page, grid columns, media info, infinite scrolling
+- **Performance Settings**: Image quality, lazy loading
 - **Video Settings**: Autoplay, mute by default, loop videos
-- **Thumbnail Management**: Generate optimized thumbnails for faster loading
+- **Thumbnail Management**: Generate optimized thumbnails for faster loading (runs on the configured worker threads; a run that stops at the per-run cap tells you how many remain)
+- **Find Duplicate Images**: Hashes every file on the worker threads with a progress bar; close the window and it keeps running, results wait in a pill
+- **Rescan Output Folder**: Drops the server-side listing cache and rewalks the folders
 
 #### **🖼️ Image Viewing Experience**
 
@@ -358,14 +390,16 @@ Use the standalone metadata viewer to analyze any ComfyUI-generated PNG:
 Import existing ComfyUI images into your database:
 
 1. **Open admin dashboard** and click **"📸 Scan Images"**
-2. **Start scan** to analyze your entire output directory
-3. **Monitor progress** with real-time status updates
+2. **Start scan** to analyze every configured output directory
+3. **Monitor progress** with real-time status updates. The scan runs on the server: closing the window only minimizes it to a progress pill, and reloading the page re-attaches to a running scan
 4. **Review results** showing:
    - Images found and processed
    - Prompts extracted and linked
    - Any errors or issues encountered
 5. **Access imported data** through the normal gallery interface
 
+The scan reads PNG text chunks and A1111-style parameters, and, when `ffprobe` is installed, the prompt ComfyUI's video savers embed in MP4/MOV/WebM files. Files are read on the number of **worker threads** set in Settings → Performance (default: half your cores), which makes a large library several times faster.
+
 ### 🏷️ AI AutoTag
 
 Automatically tag your entire image collection using AI vision models:
@@ -497,7 +531,7 @@ Keep your database optimized with built-in maintenance tools:
 
 ### Database Location
 
-By default, the database is saved as `example_prompts.db` in the node directory. This file contains all your prompts and linked images and can be backed up or shared.
+By default, the database is saved as `prompts.db` in the extension directory. It contains your prompt library and image links. See the [database path compatibility note](#database-path-compatibility-211) before upgrading a deployment with a custom path.
 
 ### Searching Prompts
 
@@ -664,7 +698,8 @@ CREATE TABLE generated_images (
 - **`utils/diagnostics.py`** - System diagnostics and health checks
 - **`web/admin.html`** - Advanced admin dashboard with metadata panel
 - **`web/index.html`** - Simple web interface
-- **`web/js/prompt_manager.js`** - Dashboard JavaScript
+- **`web/comfy/prompt_manager.js`** - ComfyUI canvas extension (the only file ComfyUI loads)
+- **`web/js/admin.js`**, **`web/js/gallery.js`** - Dashboard and gallery JavaScript
 - **`web/js/tags-page.js`** - Tag management JavaScript
 - **`web/metadata.html`** - Standalone PNG metadata viewer
 
@@ -699,8 +734,11 @@ ComfyUI_PromptManager/
 │   ├── gallery.html             # Image gallery
 │   ├── index.html               # Simple web interface
 │   ├── metadata.html            # Standalone metadata viewer
+│   ├── comfy/
+│   │   └── prompt_manager.js    # ComfyUI canvas extension (WEB_DIRECTORY)
 │   └── js/
-│       ├── prompt_manager.js    # Dashboard JavaScript
+│       ├── admin.js             # Dashboard JavaScript
+│       ├── gallery.js           # Gallery JavaScript
 │       └── tags-page.js         # Tag management JavaScript
 ├── tests/
 │   ├── __init__.py
@@ -735,6 +773,32 @@ Relative paths are resolved against the extension directory. Absolute paths are
 }
 ```
 
+### Database Path Compatibility (#211)
+
+PromptManager now honors the `database.default_path` setting in `config.json`
+when loaded as a ComfyUI custom node. Relative paths resolve against the extension
+directory, not ComfyUI's working directory; absolute paths are used as-is. Earlier
+versions could ignore this setting and use `<extension>/prompts.db` instead.
+Honoring a different configured path can therefore show an empty or different
+library. PromptManager does not automatically move or merge database files.
+
+Before upgrading or changing this setting:
+
+1. Identify the database your current installation actually uses and confirm its
+   library contents. Check both the configured path and the prior fallback
+   `<extension>/prompts.db`; keep track of which existing file you intend to use.
+2. Preserve a consistent backup of the active database with the dashboard's
+   Backup/Restore backup action or SQLite's online backup API. Do not copy only
+   the main file of a live SQLite database: committed data may still be in its
+   `-wal` file. For a filesystem copy, stop ComfyUI cleanly first and preserve any
+   remaining `-wal` file together with the main database.
+3. Set `database.default_path` to the existing intended database file. An absolute
+   path makes the selection explicit; do not move or overwrite either database
+   while choosing the path.
+4. Restart ComfyUI, then verify that the expected library appears and the active
+   database path resolves to the file you selected. Keep the backup until both
+   checks succeed.
+
 ### Gallery & Monitoring Settings
 
 Configure the automatic image detection and gallery system:
@@ -763,11 +827,15 @@ Configure the web interface behavior:
 # Web UI settings (PromptManagerConfig class)
 RESULT_TIMEOUT = 5  # Seconds to auto-hide results in ComfyUI node
 WEBUI_DISPLAY_MODE = 'newtab'  # 'popup' or 'newtab' for Web UI button
+INFINITE_SCROLL = False  # Dashboard list loads the next page on scroll by default
 SHOW_TEST_BUTTON = False  # Show API test button in node UI
 ```
 
+All of these are editable in the dashboard Settings modal and saved to `config.json` (which is private to the installation and holds the CivitAI key, so it is never committed).
+
 ### Performance Tuning
 
+- **Worker threads** (Settings → Performance): threads used by the image rescan, thumbnail generation and duplicate detection. Default is half the detected cores, maximum is the core count. Pillow releases the GIL while decoding, so the scan and thumbnails scale almost linearly; keep a few cores free while ComfyUI is generating.
 - The database automatically creates indexes for optimal search performance
 - Regular `VACUUM` operations keep the database optimized
 - Consider backing up the database periodically
@@ -802,7 +870,7 @@ print(f"Average rating: {info['average_rating']}")
 # Create backup
 db.model.backup_database("backup_prompts.db")
 
-# The database file can be copied directly for backup
+# The online backup includes committed data still in the WAL.
 ```
 
 ## Development
@@ -811,9 +879,14 @@ db.model.backup_database("backup_prompts.db")
 
 ```bash
 cd ComfyUI_PromptManager
-python -m pytest tests/ -v
+pip install -r requirements-dev.txt
+python -m unittest discover -s tests          # Python (what CI runs)
+node --test tests/js/*.test.js                 # frontend pure modules
+python -m coverage run -m unittest discover -s tests && python -m coverage combine && python -m coverage report
 ```
 
+CI runs the suite on Linux, Windows and macOS for Python 3.10 to 3.12 with an 85% coverage gate. Tests that delete a database file first check that nothing in the process still holds it open (`tests/open_handles.py`), so a connection leak fails on Linux the way it would on Windows.
+
 ### Code Style
 
 The project follows PEP 8 guidelines with:
@@ -874,6 +947,10 @@ MIT License - see LICENSE file for details.
 
 ### Recently Completed
 
+- **✅ Security & Reliability Hardening (3.3.0)**: validated gallery roots, contained file serving, safe backups and restores, background scans, 85%+ test coverage on all three operating systems
+- **✅ Worker Threads**: scan, thumbnails and duplicate detection run on a configurable thread pool
+- **✅ Tag Autocomplete & Infinite Scrolling**: faster browsing in the dashboard and gallery
+- **✅ Queue-Time Run Counting (3.2.4)**: run counts and image linking follow the ComfyUI queue
 - **✅ LoRA Manager Integration**: Import LoRA metadata, trigger words, and preview images
 - **✅ Folder Filter**: Browse and filter prompts by output subdirectory
 - **✅ Multi-Directory Gallery**: Scan multiple output directories simultaneously
@@ -893,6 +970,31 @@ MIT License - see LICENSE file for details.
 
 ## Changelog
 
+### v3.3.1 (Proxy Routing Fix)
+
+- **🌐 Proxy support**: the dashboard, gallery, tag page and metadata viewer now work when ComfyUI is reached through a proxy mount such as SwarmUI's `/ComfyBackendDirect/`. Scripts, images, API calls, progress streams and gallery folder filters stay under the mount, and the canvas node's buttons use ComfyUI's `fileURL()`. Direct ComfyUI URLs are unchanged. Thanks to @ghosttigerllc-bit ([#164](https://github.com/ComfyAssets/ComfyUI_PromptManager/pull/164))
+- **📦 Dependencies**: the runtime minimums stay at `Pillow>=8.0.0` and `watchdog>=2.1.0`, so updating Prompt Manager never forces upgrades in the shared ComfyUI environment
+
+### v3.3.0 (Hardening, Background Jobs & Worker Threads)
+
+- **🔒 Security**: gallery roots restricted to ComfyUI directories and validated wherever they are read; media-only, fail-closed image serving; Windows path escape in static routes closed; quote-safe HTML escaping with inline handlers moved to data attributes; the CivitAI key is never returned and only sent to HTTPS CivitAI hosts; auto-tag paths contained; side-effect routes are POST-only; backups include the WAL and restores are size-capped, integrity-checked and applied through SQLite's online backup API
+- **🧵 Worker Threads**: a Settings slider (default half the cores) used by the image rescan, thumbnail generation and duplicate detection. Measured: a 1 GB sample scanned in 2.5 s instead of 18 s at 8 threads
+- **🔁 Background Jobs**: the image rescan and the duplicate scan run on the server, stream progress, survive a closed tab, minimize to a progress pill and re-attach after a reload
+- **🎬 Video Prompts**: the scan reads the prompt ComfyUI's video savers embed in MP4/MOV/WebM containers (needs `ffprobe`)
+- **🖼️ Thumbnails**: written atomically and served under versioned URLs so a regenerated thumbnail never shows a stale cached copy; the per-run cap applies only to thumbnails that still need generating and the gallery says how many remain
+- **🏷️ Tag Autocomplete**: the dashboard tag box suggests tags, narrowed to those that co-occur with the tags already typed
+- **♾️ Infinite Scrolling**: optional in the gallery and the dashboard list (saved default plus a session toggle)
+- **⚙️ Gallery Settings** now apply at load; grid columns, media info and the Rescan button work as labelled
+- **🐛 Correctness**: run counts increment when ComfyUI dequeues a prompt; negative prompts are never traced; per-thread SQLite connections and idempotent migrations (an old rebuild could delete every image row); `truncate_logs` no longer rolls the log over; the ComfyUI canvas loads only the canvas extension; Windows console banners are ASCII-safe; subfolders use `/` on every platform
+- **🧪 Tests & CI**: 1,460 Python and 142 JavaScript tests (from 510 and 26), coverage 96% with an 85% gate, CI on Linux, Windows and macOS for Python 3.10–3.12, flake8 clean
+- **⚠️ API changes**: image paths in responses are relative to the ComfyUI tree (use `url`); `autotag/download`, `autotag/start` and the thumbnail progress stream are POST-only; `lora/status` returns `has_civitai_api_key` instead of the key; `/search` is paginated
+
+### v3.2.4 (Recently Used Sorting & Queue-Time Run Counts)
+
+- **🕒 Recently Used Sort**: sort the dashboard by last use, most used or rating, server-side across pages
+- **🔢 Run Counts**: `run_count` and `last_used_at` are recorded through the ComfyUI queue hook, so cached nodes still count
+- **🔗 Positive-Prompt Image Linking**: generated images link to the positive prompt traced through the queued graph
+
 ### v3.2.1 (LoRA Manager Integration)
 
 - **🔗 LoRA Manager Integration**: Import LoRA metadata, trigger words, and CivitAI example images from [ComfyUI-Lora-Manager](https://github.com/willmiao/ComfyUI-Lora-Manager) into your prompt database

__init__.py

@@ -3,11 +3,14 @@ ComfyUI_PromptManager: A comprehensive ComfyUI custom node that extends text enc
 with persistent prompt storage, advanced search capabilities, and an automatic image
 gallery system.
 
-This module provides two main node types:
+This module provides prompt node types:
 - PromptManager: CLIP encoding node that outputs CONDITIONING
-- PromptManagerText: Text-only node that outputs STRING with prepend/append functionality
+- PromptManagerFocused: Editor-first layout using the same encoding contract
+- PromptManagerSplit: Side-by-side editor and library using the same contract
+- PromptManagerText: Text-only node that outputs STRING with prepend/append
+  functionality
 
-Both nodes share the same database backend for persistent prompt storage and include
+The nodes share the same database backend for persistent prompt storage and include
 an automatic image gallery system that monitors ComfyUI output directories and links
 generated images to their source prompts.
 
@@ -37,24 +40,37 @@ def get_version():
     return "unknown"
 
 
-from .prompt_manager import PromptManager
-from .prompt_manager_text import PromptManagerText
-from .prompt_search_list import PromptSearchList
+from .prompt_manager import (  # noqa: E402
+    PromptManager,
+    PromptManagerFocused,
+    PromptManagerSplit,
+)
+from .prompt_manager_pair import PromptManagerPair  # noqa: E402
+from .prompt_manager_text import PromptManagerText  # noqa: E402
+from .prompt_search_list import PromptSearchList  # noqa: E402
 
 NODE_CLASS_MAPPINGS = {
     "PromptManager": PromptManager,
+    "PromptManagerFocused": PromptManagerFocused,
+    "PromptManagerSplit": PromptManagerSplit,
+    "PromptManagerPair": PromptManagerPair,
     "PromptManagerText": PromptManagerText,
     "PromptSearchList": PromptSearchList,
 }
 
 NODE_DISPLAY_NAME_MAPPINGS = {
-    "PromptManager": "Prompt Manager",
+    "PromptManager": "Prompt Manager (Classic)",
+    "PromptManagerFocused": "Prompt Manager (Focused)",
+    "PromptManagerSplit": "Prompt Manager (Split)",
+    "PromptManagerPair": "Prompt Manager (+/-)",
     "PromptManagerText": "Prompt Manager Text",
     "PromptSearchList": "Prompt Search List",
 }
 
-# Define path to web directory for UI components
-WEB_DIRECTORY = "web"
+# Canvas extension only. ComfyUI loads every .js under this directory into the
+# canvas page, so the admin/gallery bundles and vendored libraries under web/
+# are served by our own routes instead (see py/api).
+WEB_DIRECTORY = "web/comfy"
 
 # Add API routes (same pattern as ComfyUI_Assets)
 try:
@@ -71,6 +87,9 @@ try:
     routes = config.routes
     api = PromptManagerAPI()
     api.add_routes(routes)
+    host_app = getattr(config.server_instance, "app", None)
+    if host_app is not None and hasattr(host_app, "on_cleanup"):
+        host_app.on_cleanup.append(api.cleanup_history_reads)
 
 except Exception as e:
     # Log to internal logging system without console spam
@@ -82,6 +101,10 @@ except Exception as e:
     except Exception:
         pass
 
+# Shared database, set up with the monitoring system below. Stays None when that
+# fails so the queue hook can report the database as unavailable instead of crashing.
+_global_db = None
+
 # Start image monitoring globally at module import time
 # This ensures images are linked regardless of which node is used in the workflow
 try:
@@ -111,12 +134,69 @@ except Exception as e:
     except Exception:
         print(f"[ComfyUI-PromptManager] Warning: Failed to start image monitoring: {e}")
 
+
+def _db_factory():
+    """The shared database for the queue hook; called lazily per request."""
+    if _global_db is None:
+        raise RuntimeError("PromptManager database unavailable")
+    return _global_db
+
+
+# Count prompt runs when ComfyUI dequeues workflows (it skips unchanged nodes, so
+# node execution alone misses re-runs). Separate from monitoring so either can fail.
+try:
+    from .py.config import server_instance
+    from .utils.usage_tracking import register_queue_hook
+
+    register_queue_hook(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 usage hook: {e}")
+    except Exception:
+        import logging
+
+        logging.getLogger("prompt_manager.init").error(
+            f"Failed to register usage hook: {e}"
+        )
+
+
+def _history_db_factory() -> "PromptDatabase":
+    """Initialize history on its worker even if monitoring initialization failed."""
+    if _global_db is not None:
+        return _global_db
+    from .database.operations import PromptDatabase
+
+    return PromptDatabase()
+
+
+_history_registration = None
+try:
+    from .py.config import server_instance
+    from .utils.execution_history import register_history_hook
+
+    _history_registration = register_history_hook(server_instance, _history_db_factory)
+except Exception:
+    try:
+        from .utils.logging_config import get_logger
+
+        get_logger("prompt_manager.init").error("Failed to register execution history")
+    except Exception:
+        import logging
+
+        logging.getLogger("prompt_manager.init").error(
+            "Failed to register execution history"
+        )
+
 __all__ = ["NODE_CLASS_MAPPINGS", "NODE_DISPLAY_NAME_MAPPINGS", "WEB_DIRECTORY"]
 
 # Print startup message with loaded tools
 print()
 print(f"\033[94m[ComfyUI-PromptManager] Version:\033[0m {get_version()}")
 for node_key, display_name in NODE_DISPLAY_NAME_MAPPINGS.items():
-    print(f"🫶 \033[94mLoaded:\033[0m {display_name}")
+    # ASCII only: a Windows console using a legacy code page cannot print emoji,
+    # and a UnicodeEncodeError here would abort loading the whole node.
+    print(f"  \033[94mLoaded:\033[0m {display_name}")
 print(f"\033[94mTotal: {len(NODE_CLASS_MAPPINGS)} tools loaded\033[0m")
 print()

database/execution_context_schema.py

@@ -0,0 +1,406 @@
+"""Owned execution-context v1 tables within execution-history schema v2.
+
+Content IDs hash exact scalar values; evidence JSON contains metadata only.
+All validation scans use cursors and bounded per-row payloads.
+"""
+
+import json
+import sqlite3
+
+from .execution_history_schema import _compact, _uuid_check
+
+try:
+    from ..utils.execution_contexts import (
+        BUNDLE_STATES,
+        CAPTURE_REASONS,
+        EVIDENCE_SOURCES,
+        UNKNOWN_REASONS,
+        MAX_CONTEXT_BYTES,
+        ContextBundle,
+        PromptSnapshot,
+        RecipeContext,
+        TextEvidence,
+        validate_bundle,
+        content_digest,
+        text_size,
+    )
+except ImportError:
+    from utils.execution_contexts import (
+        BUNDLE_STATES,
+        CAPTURE_REASONS,
+        EVIDENCE_SOURCES,
+        UNKNOWN_REASONS,
+        MAX_CONTEXT_BYTES,
+        ContextBundle,
+        PromptSnapshot,
+        RecipeContext,
+        TextEvidence,
+        validate_bundle,
+        content_digest,
+        text_size,
+    )
+
+
+FIELDS = ("raw", "effective", "prepend", "append")
+TABLE_COLUMNS = {
+    "execution_context_texts": (
+        ("text_id", "TEXT", 1, 1),
+        ("text_value", "TEXT", 1, 0),
+    ),
+    "execution_prompt_snapshots": (
+        ("snapshot_id", "TEXT", 1, 1),
+        *((f"{field}_text_id", "TEXT", 0, 0) for field in FIELDS),
+    ),
+    "execution_contexts": (
+        ("context_id", "TEXT", 1, 1),
+        ("positive_snapshot_id", "TEXT", 0, 0),
+        ("negative_snapshot_id", "TEXT", 0, 0),
+    ),
+    "execution_run_contexts": (
+        ("run_id", "TEXT", 1, 1),
+        ("anchor_id", "TEXT", 1, 2),
+        ("context_id", "TEXT", 1, 0),
+        ("evidence_json", "TEXT", 1, 0),
+        ("positive_prompt_id", "INTEGER", 0, 0),
+        ("negative_prompt_id", "INTEGER", 0, 0),
+        ("pair_id", "INTEGER", 0, 0),
+        ("reference_epoch", "TEXT", 0, 0),
+        ("capture_version", "INTEGER", 1, 0),
+    ),
+    "execution_context_capture": (
+        ("run_id", "TEXT", 1, 1),
+        ("state", "TEXT", 1, 0),
+        ("reason", "TEXT", 1, 0),
+        ("format_version", "INTEGER", 1, 0),
+    ),
+}
+
+
+def _digest_check(column: str, nullable: bool = False) -> str:
+    check = (
+        f"typeof({column}) = 'text' AND length({column}) = 64 "
+        f"AND length(CAST({column} AS BLOB)) = 64 "
+        f"AND {column} NOT GLOB '*[^0-9a-f]*'"
+    )
+    return f"{column} IS NULL OR ({check})" if nullable else check
+
+
+def _bounded_uuid_check(column: str) -> str:
+    return f"{_uuid_check(column)} AND length(CAST({column} AS BLOB)) = 36"
+
+
+def _enum_check(column: str, values: frozenset[str]) -> str:
+    return (
+        f"typeof({column}) = 'text' AND {column} IN ("
+        + ",".join("'" + value + "'" for value in sorted(values))
+        + ")"
+    )
+
+
+TABLE_CHECKS = {
+    "execution_context_texts": (
+        _digest_check("text_id"),
+        "typeof(text_value) = 'text' AND length(CAST(text_value AS BLOB)) <= 65536",
+    ),
+    "execution_prompt_snapshots": (
+        _digest_check("snapshot_id"),
+        *(_digest_check(f"{field}_text_id", True) for field in FIELDS),
+    ),
+    "execution_contexts": (
+        _digest_check("context_id"),
+        _digest_check("positive_snapshot_id", True),
+        _digest_check("negative_snapshot_id", True),
+        "positive_snapshot_id IS NOT NULL OR negative_snapshot_id IS NOT NULL",
+    ),
+    "execution_run_contexts": (
+        _bounded_uuid_check("run_id"),
+        "typeof(anchor_id) = 'text' AND length(anchor_id) BETWEEN 1 AND 128 "
+        "AND length(CAST(anchor_id AS BLOB)) <= 512",
+        _digest_check("context_id"),
+        "typeof(evidence_json) = 'text' AND length(CAST(evidence_json AS "
+        "BLOB)) <= 8192",
+        *(
+            f"{column} IS NULL OR (typeof({column}) = 'integer' AND {column} > 0)"
+            for column in ("positive_prompt_id", "negative_prompt_id", "pair_id")
+        ),
+        f"reference_epoch IS NULL OR ({_bounded_uuid_check('reference_epoch')})",
+        "typeof(capture_version) = 'integer' AND capture_version = 1",
+    ),
+    "execution_context_capture": (
+        _bounded_uuid_check("run_id"),
+        _enum_check("state", BUNDLE_STATES),
+        _enum_check("reason", CAPTURE_REASONS),
+        "length(CAST(reason AS BLOB)) <= 64",
+        "typeof(format_version) = 'integer' AND format_version = 1",
+    ),
+}
+# (source column, destination table, destination column); actions always NO ACTION.
+TABLE_FOREIGN_KEYS = {
+    "execution_context_texts": (),
+    "execution_prompt_snapshots": tuple(
+        (f"{field}_text_id", "execution_context_texts", "text_id") for field in FIELDS
+    ),
+    "execution_contexts": tuple(
+        (f"{role}_snapshot_id", "execution_prompt_snapshots", "snapshot_id")
+        for role in ("positive", "negative")
+    ),
+    "execution_run_contexts": (
+        ("run_id", "execution_runs", "run_id"),
+        ("context_id", "execution_contexts", "context_id"),
+    ),
+    "execution_context_capture": (("run_id", "execution_runs", "run_id"),),
+}
+# Only SQLite-generated primary-key indexes are unique. No secondary indexes required.
+
+
+def install_context_schema(conn: sqlite3.Connection) -> None:
+    """Install tables inside the caller's transaction, without committing."""
+    for table, columns in TABLE_COLUMNS.items():
+        definitions = [
+            f"{name} {kind}{' NOT NULL' if required else ''}"
+            for name, kind, required, _ in columns
+        ]
+        primary = [name for name, _, _, order in columns if order]
+        definitions.append(f"PRIMARY KEY ({', '.join(primary)})")
+        definitions.extend(f"CHECK({check})" for check in TABLE_CHECKS[table])
+        definitions.extend(
+            f"FOREIGN KEY ({column}) REFERENCES {target}({key}) "
+            "ON UPDATE NO ACTION ON DELETE NO ACTION"
+            for column, target, key in TABLE_FOREIGN_KEYS[table]
+        )
+        conn.execute(f"CREATE TABLE {table} ({', '.join(definitions)})")
+
+
+def _unique_object(pairs: list[tuple[str, object]]) -> dict:
+    result = {}
+    for key, value in pairs:
+        if key in result:
+            raise ValueError("Duplicate evidence key")
+        result[key] = value
+    return result
+
+
+def _snapshot_values(conn: sqlite3.Connection, snapshot_id: str) -> tuple:
+    row = conn.execute(
+        "SELECT raw_text_id,effective_text_id,prepend_text_id,append_text_id "
+        "FROM execution_prompt_snapshots WHERE snapshot_id=?",
+        (snapshot_id,),
+    ).fetchone()
+    if row is None:
+        raise ValueError("Missing snapshot")
+    values = []
+    for text_id in row:
+        if text_id is None:
+            values.append(None)
+        else:
+            text = conn.execute(
+                "SELECT text_value FROM execution_context_texts WHERE text_id=?",
+                (text_id,),
+            ).fetchone()
+            if text is None:
+                raise ValueError("Missing context text")
+            text_size(text[0])
+            values.append(text[0])
+    return tuple(values)
+
+
+def _validate_evidence(
+    conn: sqlite3.Connection, anchor_id: str, context_id: str, encoded: str
+) -> tuple[RecipeContext, bool]:
+    text_size(encoded, 8192)
+    evidence = json.loads(encoded, object_pairs_hook=_unique_object)
+    if (
+        type(evidence) is not dict
+        or set(evidence)
+        != {"format_version", "role_provenance", "positive", "negative"}
+        or type(evidence["format_version"]) is not int
+        or evidence["format_version"] != 1
+        or evidence["role_provenance"] != "inferred"
+    ):
+        raise ValueError("Invalid context evidence")
+    row = conn.execute(
+        "SELECT positive_snapshot_id,negative_snapshot_id FROM "
+        "execution_contexts WHERE context_id=?",
+        (context_id,),
+    ).fetchone()
+    if row is None:
+        raise ValueError("Missing context")
+    complete = True
+    snapshots = []
+    for role, snapshot_id in zip(("positive", "negative"), row):
+        metadata = evidence[role]
+        if snapshot_id is None:
+            if metadata is not None:
+                raise ValueError("Unexpected role evidence")
+            complete = False
+            snapshots.append(None)
+            continue
+        values = _snapshot_values(conn, snapshot_id)
+        if type(metadata) is not dict or set(metadata) != set(FIELDS):
+            raise ValueError("Invalid snapshot evidence")
+        fields = []
+        for field, value in zip(FIELDS, values):
+            item = metadata[field]
+            if type(item) is not dict or set(item) != {
+                "provenance",
+                "source",
+                "reason",
+            }:
+                raise ValueError("Invalid field evidence")
+            provenance, source, reason = (
+                item["provenance"],
+                item["source"],
+                item["reason"],
+            )
+            if (
+                type(provenance) is not str
+                or type(source) is not str
+                or source not in EVIDENCE_SOURCES.get(provenance, ())
+            ):
+                raise ValueError("Invalid evidence source")
+            if value is None:
+                complete = False
+                if (
+                    provenance != "unknown"
+                    or type(reason) is not str
+                    or reason not in UNKNOWN_REASONS
+                ):
+                    raise ValueError("Unknown evidence requires reason")
+            elif provenance == "unknown" or reason is not None:
+                raise ValueError("Known evidence metadata mismatch")
+            fields.append(TextEvidence(value, provenance, source, reason))
+        snapshots.append(PromptSnapshot(*fields))
+    return RecipeContext(anchor_id, *snapshots, evidence["role_provenance"]), complete
+
+
+def context_schema_error(conn: sqlite3.Connection) -> str | None:
+    """Validate exact owned schema, bounded evidence, fingerprints and links."""
+    if conn.execute(
+        "SELECT 1 FROM sqlite_master WHERE type IN ('trigger','view') LIMIT 1"
+    ).fetchone():
+        return "Unsupported execution context trigger/view"
+    tables = {
+        row[0]: row[1]
+        for row in conn.execute("SELECT name,sql FROM sqlite_master WHERE type='table'")
+        if row[0] in TABLE_COLUMNS
+    }
+    if tables.keys() != TABLE_COLUMNS.keys():
+        return "Incomplete execution context schema"
+    for table, expected in TABLE_COLUMNS.items():
+        columns = tuple(
+            (row[1], row[2].upper(), row[3], row[5])
+            for row in conn.execute(f"PRAGMA table_xinfo({table})")
+        )
+        if columns != expected:
+            return f"{table} has invalid execution context columns"
+        sql = _compact(tables[table])
+        checks = TABLE_CHECKS[table]
+        if sql.count("check(") != len(checks) or any(
+            f"check({_compact(check)})" not in sql for check in checks
+        ):
+            return f"{table} has invalid execution context checks"
+        foreign = tuple(
+            sorted(
+                (row[3], row[2], row[4], row[5], row[6], row[7])
+                for row in conn.execute(f"PRAGMA foreign_key_list({table})")
+            )
+        )
+        expected_foreign = tuple(
+            sorted(
+                (*key, "NO ACTION", "NO ACTION", "NONE")
+                for key in TABLE_FOREIGN_KEYS[table]
+            )
+        )
+        if foreign != expected_foreign:
+            return f"{table} has invalid execution context foreign keys"
+        for index in conn.execute(f"PRAGMA index_list({table})"):
+            if not index[2]:
+                continue
+            name = index[1].replace('"', '""')
+            keys = tuple(
+                row for row in conn.execute(f'PRAGMA index_xinfo("{name}")') if row[5]
+            )
+            primary = tuple(column[0] for column in expected if column[3])
+            if (
+                index[3] != "pk"
+                or index[4]
+                or tuple(key[2] for key in keys) != primary
+                or any(key[3] or key[4] != "BINARY" for key in keys)
+            ):
+                return f"{table} has unexpected execution context uniqueness"
+        predicate = " AND ".join(f"({check})" for check in checks)
+        if conn.execute(
+            f"SELECT 1 FROM {table} WHERE NOT ({predicate}) LIMIT 1"
+        ).fetchone():
+            return f"{table} contains invalid execution context records"
+        if conn.execute(f"PRAGMA foreign_key_check({table})").fetchone():
+            return f"{table} contains orphan execution context records"
+    if conn.execute(
+        "SELECT 1 FROM execution_run_contexts GROUP BY run_id HAVING "
+        "COUNT(*) > 16 LIMIT 1"
+    ).fetchone():
+        return "Too many execution contexts per run"
+    try:
+        for text_id, text in conn.execute(
+            "SELECT text_id,text_value FROM execution_context_texts"
+        ):
+            text_size(text)
+            if content_digest("text", (text,)) != text_id:
+                raise ValueError("Invalid text fingerprint")
+        for (snapshot_id,) in conn.execute(
+            "SELECT snapshot_id FROM execution_prompt_snapshots"
+        ):
+            if (
+                content_digest("snapshot", _snapshot_values(conn, snapshot_id))
+                != snapshot_id
+            ):
+                raise ValueError("Invalid snapshot fingerprint")
+        for context_id, positive, negative in conn.execute(
+            "SELECT context_id,positive_snapshot_id,negative_snapshot_id FROM "
+            "execution_contexts"
+        ):
+            if content_digest("context", (positive, negative)) != context_id:
+                raise ValueError("Invalid context fingerprint")
+            payload_size = 0
+            for snapshot_id in (positive, negative):
+                if snapshot_id is not None:
+                    payload_size += sum(
+                        text_size(value)
+                        for value in _snapshot_values(conn, snapshot_id)
+                        if value is not None
+                    )
+            if payload_size > MAX_CONTEXT_BYTES:
+                raise ValueError("Oversized detached context")
+        previous_run = None
+        contexts = ()
+        for run_id, anchor_id, context_id, evidence in conn.execute(
+            "SELECT run_id,anchor_id,context_id,evidence_json FROM "
+            "execution_run_contexts ORDER BY run_id,anchor_id"
+        ):
+            if run_id != previous_run:
+                contexts = ()
+                previous_run = run_id
+            context, complete = _validate_evidence(
+                conn, anchor_id, context_id, evidence
+            )
+            status = conn.execute(
+                "SELECT state,reason FROM execution_context_capture WHERE run_id=?",
+                (run_id,),
+            ).fetchone()
+            if status is None:
+                raise ValueError("Context link has no capture header")
+            state, reason = status
+            if state == "complete" and not complete:
+                raise ValueError("Incomplete evidence marked complete")
+            candidate = (*contexts, context)
+            validate_bundle(ContextBundle(candidate, state, reason))
+            contexts = candidate
+        if conn.execute(
+            "SELECT 1 FROM execution_context_capture c WHERE state='complete' "
+            "AND NOT EXISTS (SELECT 1 FROM execution_run_contexts r WHERE "
+            "r.run_id=c.run_id) LIMIT 1"
+        ).fetchone():
+            raise ValueError("Complete capture has no contexts")
+    except (ValueError, TypeError, RecursionError, OverflowError) as exc:
+        return f"Invalid execution context content: {type(exc).__name__}"
+    return None

database/execution_contexts.py

@@ -0,0 +1,411 @@
+"""Immutable execution capture and exact, epoch-fenced library navigation hints."""
+
+from contextlib import ExitStack
+import json
+import sqlite3
+from typing import Callable, ContextManager, Iterator
+
+try:
+    from ..utils.execution_history_events import RunFinish, encode_command
+    from ..utils.execution_contexts import (
+        CAPTURE_REASONS,
+        ContextBundle,
+        content_digest,
+        validate_bundle,
+    )
+except ImportError:
+    from utils.execution_history_events import RunFinish, encode_command
+    from utils.execution_contexts import (
+        CAPTURE_REASONS,
+        ContextBundle,
+        content_digest,
+        validate_bundle,
+    )
+
+
+FIELDS = ("raw", "effective", "prepend", "append")
+ROLES = ("positive", "negative")
+
+
+def _prepare(bundle: ContextBundle) -> list[dict]:
+    validate_bundle(bundle)
+    if bundle.reason not in CAPTURE_REASONS:
+        raise ValueError("Invalid capture reason")
+    prepared = []
+    complete = bool(bundle.contexts)
+    for context in bundle.contexts:
+        evidence = {"format_version": 1, "role_provenance": context.role_provenance}
+        snapshots = []
+        for role in ROLES:
+            snapshot = getattr(context, role)
+            if snapshot is None:
+                snapshots.append(None)
+                evidence[role] = None
+                complete = False
+                continue
+            values = tuple(getattr(snapshot, field).value for field in FIELDS)
+            complete = complete and all(value is not None for value in values)
+            snapshots.append((content_digest("snapshot", values), values))
+            evidence[role] = {
+                field: {
+                    "provenance": getattr(snapshot, field).provenance,
+                    "source": getattr(snapshot, field).source,
+                    "reason": getattr(snapshot, field).reason,
+                }
+                for field in FIELDS
+            }
+        if all(snapshot is None for snapshot in snapshots):
+            raise ValueError("Context must contain a snapshot")
+        context_id = content_digest(
+            "context", tuple(s[0] if s else None for s in snapshots)
+        )
+        encoded = json.dumps(
+            evidence,
+            ensure_ascii=False,
+            allow_nan=False,
+            separators=(",", ":"),
+            sort_keys=True,
+        )
+        if len(encoded.encode("utf-8")) > 8192:
+            raise ValueError("Oversized context evidence")
+        prepared.append(
+            {
+                "anchor_id": context.anchor_id,
+                "context_id": context_id,
+                "snapshots": snapshots,
+                "evidence_json": encoded,
+            }
+        )
+    if bundle.state == "complete" and not complete:
+        raise ValueError("Incomplete scalar capture")
+    return sorted(prepared, key=lambda item: item["anchor_id"])
+
+
+def _existing_row(
+    conn: sqlite3.Connection, table: str, columns: tuple[str, ...], values: tuple
+) -> bool:
+    row = conn.execute(
+        f"SELECT {','.join(columns[1:])} FROM {table} WHERE {columns[0]}=?",
+        (values[0],),
+    ).fetchone()
+    if row is None:
+        return False
+    if tuple(row) != values[1:]:
+        raise ValueError("Execution context digest collision")
+    return True
+
+
+def _content_rows(item: dict) -> Iterator[tuple[str, tuple[str, ...], tuple]]:
+    snapshot_ids = []
+    for snapshot in item["snapshots"]:
+        if snapshot is None:
+            snapshot_ids.append(None)
+            continue
+        snapshot_id, values = snapshot
+        text_ids = []
+        for value in values:
+            text_id = None if value is None else content_digest("text", (value,))
+            if value is not None:
+                yield "execution_context_texts", ("text_id", "text_value"), (
+                    text_id,
+                    value,
+                )
+            text_ids.append(text_id)
+        yield "execution_prompt_snapshots", (
+            "snapshot_id",
+            *(f"{field}_text_id" for field in FIELDS),
+        ), (snapshot_id, *text_ids)
+        snapshot_ids.append(snapshot_id)
+    yield "execution_contexts", (
+        "context_id",
+        "positive_snapshot_id",
+        "negative_snapshot_id",
+    ), (item["context_id"], *snapshot_ids)
+
+
+def _insert_exact(
+    conn: sqlite3.Connection, table: str, columns: tuple[str, ...], values: tuple
+) -> None:
+    if not _existing_row(conn, table, columns, values):
+        names = ",".join(columns)
+        placeholders = ",".join("?" for _ in columns)
+        conn.execute(f"INSERT INTO {table} ({names}) VALUES ({placeholders})", values)
+
+
+def _role_hints(role: str) -> str:
+    return (
+        f"SELECT r.{role}_prompt_id, tx.text_value "
+        "FROM execution_run_contexts r "
+        "JOIN execution_contexts c ON c.context_id = r.context_id "
+        f"JOIN execution_prompt_snapshots s ON s.snapshot_id = c.{role}_snapshot_id "
+        "JOIN execution_context_texts tx ON tx.text_id = s.raw_text_id "
+        f"WHERE r.{role}_prompt_id IS NOT NULL AND r.reference_epoch = ("
+        "SELECT storage_epoch FROM execution_history_metadata WHERE singleton_id = 1)"
+    )
+
+
+# Set form of the read-time prompt-hint rule in get_execution_contexts():
+# a stored role hint is live only in the current storage epoch, and only when
+# its exact raw snapshot text matches exactly one prompt, which is the hinted
+# one (_prompt_hint). Evaluated once per statement, never per prompt row.
+# Defines the CTE ``live_hint_prompts(prompt_id)``; prefix it with WITH.
+LIVE_HINT_CTES = (
+    "stored_hints(prompt_id, raw) AS ("
+    + _role_hints("positive")
+    + " UNION "
+    + _role_hints("negative")
+    + "), unique_hint_texts(prompt_id, raw) AS ("
+    "SELECT MIN(id), text COLLATE BINARY FROM prompts "
+    "WHERE text COLLATE BINARY IN (SELECT raw FROM stored_hints) "
+    "GROUP BY text COLLATE BINARY HAVING COUNT(*) = 1"
+    "), live_hint_prompts(prompt_id) AS ("
+    "SELECT h.prompt_id FROM stored_hints h JOIN unique_hint_texts u "
+    "ON u.prompt_id = h.prompt_id AND u.raw = h.raw COLLATE BINARY)"
+)
+LIVE_HINT_PROMPT_IDS_SQL = (
+    "WITH " + LIVE_HINT_CTES + " SELECT DISTINCT prompt_id FROM live_hint_prompts"
+)
+
+
+def _prompt_hint(conn: sqlite3.Connection, raw: str | None) -> int | None:
+    if raw is None:
+        return None
+    rows = conn.execute(
+        "SELECT id FROM prompts WHERE text=? COLLATE BINARY LIMIT 2", (raw,)
+    ).fetchall()
+    return rows[0][0] if len(rows) == 1 else None
+
+
+def _hints(
+    conn: sqlite3.Connection, positive: str | None, negative: str | None
+) -> tuple:
+    pos_id, neg_id = _prompt_hint(conn, positive), _prompt_hint(conn, negative)
+    pair_id = None
+    negative_ambiguous = (
+        negative is not None
+        and neg_id is None
+        and conn.execute(
+            "SELECT 1 FROM prompts WHERE text=? COLLATE BINARY LIMIT 1", (negative,)
+        ).fetchone()
+        is not None
+    )
+    if pos_id is not None and negative is not None and not negative_ambiguous:
+        rows = conn.execute(
+            "SELECT pp.id FROM prompt_pairs pp LEFT JOIN prompts n ON "
+            "n.id=pp.negative_prompt_id "
+            "WHERE pp.positive_prompt_id=? AND ((pp.negative_prompt_id IS NULL AND "
+            "?='') "
+            "OR (pp.negative_prompt_id=? AND n.text=? COLLATE BINARY)) LIMIT 2",
+            (pos_id, negative, neg_id, negative),
+        ).fetchall()
+        if len(rows) == 1:
+            pair_id = rows[0][0]
+    return pos_id, neg_id, pair_id
+
+
+def _values(conn: sqlite3.Connection, snapshot_id: str | None) -> dict | None:
+    if snapshot_id is None:
+        return None
+    row = conn.execute(
+        "SELECT r.text_value,e.text_value,p.text_value,a.text_value "
+        "FROM execution_prompt_snapshots s "
+        "LEFT JOIN execution_context_texts r ON r.text_id=s.raw_text_id "
+        "LEFT JOIN execution_context_texts e ON e.text_id=s.effective_text_id "
+        "LEFT JOIN execution_context_texts p ON p.text_id=s.prepend_text_id "
+        "LEFT JOIN execution_context_texts a ON a.text_id=s.append_text_id "
+        "WHERE s.snapshot_id=?",
+        (snapshot_id,),
+    ).fetchone()
+    if row is None:
+        raise sqlite3.DatabaseError("Missing execution context snapshot")
+    return dict(zip(FIELDS, row))
+
+
+def _write_capture(
+    conn: sqlite3.Connection,
+    run_id: str,
+    expected_epoch: str,
+    bundle: ContextBundle,
+    prepared: list[dict],
+) -> bool:
+    capture = conn.execute(
+        "SELECT state,reason,format_version FROM execution_context_capture "
+        "WHERE run_id=?",
+        (run_id,),
+    ).fetchone()
+    if capture is not None:
+        for item in prepared:
+            for table, columns, values in _content_rows(item):
+                _existing_row(conn, table, columns, values)
+        links = conn.execute(
+            "SELECT anchor_id,context_id,evidence_json FROM "
+            "execution_run_contexts WHERE run_id=? ORDER BY anchor_id",
+            (run_id,),
+        ).fetchall()
+        expected = [
+            (item["anchor_id"], item["context_id"], item["evidence_json"])
+            for item in prepared
+        ]
+        # Compare JSON objects so serialization order never creates a conflict.
+        original = [(row[0], row[1], json.loads(row[2])) for row in links]
+        proposed = [
+            (anchor, context, json.loads(evidence))
+            for anchor, context, evidence in expected
+        ]
+        if capture[0] != "conflicting" and (
+            tuple(capture) != (bundle.state, bundle.reason, bundle.format_version)
+            or original != proposed
+        ):
+            conn.execute(
+                "UPDATE execution_context_capture SET "
+                "state='conflicting',reason='conflicting' WHERE run_id=?",
+                (run_id,),
+            )
+        return True
+    for item in prepared:
+        for table, columns, values in _content_rows(item):
+            _insert_exact(conn, table, columns, values)
+        raw = [snapshot[1][0] if snapshot else None for snapshot in item["snapshots"]]
+        positive_id, negative_id, pair_id = _hints(conn, *raw)
+        conn.execute(
+            "INSERT INTO "
+            "execution_run_contexts(run_id,anchor_id,context_id,evidence_json,"
+            "positive_prompt_id,negative_prompt_id,pair_id,reference_epoch,"
+            "capture_version) VALUES(?,?,?,?,?,?,?,?,1)",
+            (
+                run_id,
+                item["anchor_id"],
+                item["context_id"],
+                item["evidence_json"],
+                positive_id,
+                negative_id,
+                pair_id,
+                expected_epoch,
+            ),
+        )
+    conn.execute(
+        "INSERT INTO "
+        "execution_context_capture(run_id,state,reason,format_version) "
+        "VALUES(?,?,?,?)",
+        (run_id, bundle.state, bundle.reason, bundle.format_version),
+    )
+    return True
+
+
+class ExecutionContextOperations:
+    """Compose with ExecutionHistoryOperations on PromptDatabase."""
+
+    def write_execution_contexts(
+        self, run_id: str, expected_epoch: str, bundle: ContextBundle
+    ) -> bool:
+        prepared = _prepare(bundle)
+        with self._execution_transaction(expected_epoch) as conn:
+            if (
+                conn.execute(
+                    "SELECT 1 FROM execution_runs WHERE run_id=? AND storage_epoch=?",
+                    (run_id, expected_epoch),
+                ).fetchone()
+                is None
+            ):
+                return False
+            return _write_capture(conn, run_id, expected_epoch, bundle, prepared)
+
+    def write_execution_contexts_for_finish(
+        self,
+        command: RunFinish,
+        bundle: ContextBundle,
+        *,
+        generation_guard: Callable[[], ContextManager[bool]] | None = None,
+    ) -> bool:
+        encode_command(command)
+        if type(command) is not RunFinish:
+            raise ValueError("Expected execution finish command")
+        prepared = _prepare(bundle)
+        # Enter the guard after BEGIN, and keep it until COMMIT/ROLLBACK exits.
+        with ExitStack() as guards:
+            with self._execution_transaction(command.storage_epoch) as conn:
+                if generation_guard is not None and not guards.enter_context(
+                    generation_guard()
+                ):
+                    return False
+                row = conn.execute(
+                    "SELECT run_id FROM execution_runs WHERE session_id=? "
+                    "AND queue_item_id=? AND storage_epoch=?",
+                    (command.session_id, command.queue_item_id, command.storage_epoch),
+                ).fetchone()
+                if row is None:
+                    return False
+                return _write_capture(
+                    conn, row[0], command.storage_epoch, bundle, prepared
+                )
+
+    def get_execution_contexts(self, run_id: str) -> dict:
+        """Return first-capture scalars; hints are None unless currently valid.
+
+        reference_epoch remains historical evidence even when hints are invalidated.
+        Reads never rebind a first-capture reference or alter stored capture data.
+        """
+        conn = self.model.get_connection()
+        own_transaction = not conn.in_transaction
+        if own_transaction:
+            conn.execute("BEGIN")
+        try:
+            capture = conn.execute(
+                "SELECT state,reason,format_version FROM execution_context_capture "
+                "WHERE run_id=?",
+                (run_id,),
+            ).fetchone()
+            if capture is None:
+                return {
+                    "state": "not_captured",
+                    "reason": "missing",
+                    "format_version": None,
+                    "contexts": [],
+                }
+            epoch = self.get_execution_epoch()
+            contexts = []
+            rows = conn.execute(
+                "SELECT r.*,c.positive_snapshot_id,c.negative_snapshot_id FROM "
+                "execution_run_contexts r JOIN execution_contexts c ON "
+                "c.context_id=r.context_id WHERE r.run_id=? ORDER BY r.anchor_id",
+                (run_id,),
+            )
+            for row in rows:
+                positive = _values(conn, row["positive_snapshot_id"])
+                negative = _values(conn, row["negative_snapshot_id"])
+                hints = (None, None, None)
+                if row["reference_epoch"] == epoch:
+                    current = _hints(
+                        conn,
+                        positive["raw"] if positive else None,
+                        negative["raw"] if negative else None,
+                    )
+                    hints = tuple(
+                        row[key] if row[key] == value else None
+                        for key, value in zip(
+                            ("positive_prompt_id", "negative_prompt_id", "pair_id"),
+                            current,
+                        )
+                    )
+                contexts.append(
+                    {
+                        "anchor_id": row["anchor_id"],
+                        "context_id": row["context_id"],
+                        "positive": positive,
+                        "negative": negative,
+                        "evidence": json.loads(row["evidence_json"]),
+                        "positive_prompt_id": hints[0],
+                        "negative_prompt_id": hints[1],
+                        "pair_id": hints[2],
+                        "reference_epoch": row["reference_epoch"],
+                    }
+                )
+            return {
+                "state": capture[0],
+                "reason": capture[1],
+                "format_version": capture[2],
+                "contexts": contexts,
+            }
+        finally:
+            if own_transaction:
+                conn.rollback()

database/execution_history.py

@@ -0,0 +1,198 @@
+"""Transactional attempt writes and process-local restore invalidation."""
+
+import os
+import sqlite3
+import threading
+from contextlib import contextmanager
+from typing import Callable, Iterator
+
+try:
+    from ..utils.execution_history_events import (
+        RunFinish,
+        RunStart,
+        classify_evidence,
+        encode_command,
+    )
+    from ..utils.logging_config import get_logger
+except ImportError:
+    from utils.execution_history_events import (
+        RunFinish,
+        RunStart,
+        classify_evidence,
+        encode_command,
+    )
+    from utils.logging_config import get_logger
+
+
+_logger = get_logger("prompt_manager.database.execution_history")
+# Bounded wait for a history write lock. History is best effort and must
+# never stall ComfyUI's queue: a write that cannot start in time raises
+# "database is locked" and the writer records a gap instead of waiting.
+EXECUTION_BUSY_TIMEOUT_MS = 250
+_restore_lock = threading.Lock()
+_restore_observers: dict[str, dict[object, Callable[[], None]]] = {}
+
+
+def _database_identity(db_path: str) -> str:
+    return os.path.normcase(os.path.realpath(db_path))
+
+
+def subscribe_history_restore(
+    db_path: str, callback: Callable[[], None]
+) -> Callable[[], None]:
+    """Observe successful restores of the canonical database path."""
+    identity = _database_identity(db_path)
+    token = object()
+    with _restore_lock:
+        _restore_observers.setdefault(identity, {})[token] = callback
+
+    def unsubscribe() -> None:
+        with _restore_lock:
+            observers = _restore_observers.get(identity)
+            if observers is not None:
+                observers.pop(token, None)
+                if not observers:
+                    _restore_observers.pop(identity, None)
+
+    return unsubscribe
+
+
+def notify_history_restore(db_path: str) -> None:
+    """Notify outside the registry lock; observers cannot fail a restore."""
+    with _restore_lock:
+        callbacks = tuple(
+            _restore_observers.get(_database_identity(db_path), {}).values()
+        )
+    for callback in callbacks:
+        try:
+            callback()
+        except Exception:
+            _logger.warning("Execution history restore observer failed", exc_info=True)
+
+
+class HistoryEpochChanged(Exception):
+    """A command belongs to storage replaced by a completed restore."""
+
+    def __init__(self, storage_epoch: str):
+        self.storage_epoch = storage_epoch
+        super().__init__("Execution history storage epoch changed")
+
+
+class ExecutionHistoryOperations:
+    """Mixin using the calling thread's existing managed model connection."""
+
+    def get_execution_epoch(self) -> str:
+        row = (
+            self.model.get_connection()
+            .execute(
+                "SELECT storage_epoch FROM execution_history_metadata "
+                "WHERE singleton_id=1"
+            )
+            .fetchone()
+        )
+        if row is None:
+            raise sqlite3.DatabaseError("Missing execution history metadata")
+        return row[0]
+
+    @contextmanager
+    def _execution_transaction(
+        self, expected_epoch: str
+    ) -> Iterator[sqlite3.Connection]:
+        conn = self.model.get_connection()
+        if conn.in_transaction:
+            raise sqlite3.ProgrammingError(
+                "Execution history requires its own transaction"
+            )
+        timeout = conn.execute("PRAGMA busy_timeout").fetchone()[0]
+        conn.execute(f"PRAGMA busy_timeout={int(EXECUTION_BUSY_TIMEOUT_MS)}")
+        try:
+            conn.execute("BEGIN IMMEDIATE")
+            try:
+                epoch = self.get_execution_epoch()
+                if epoch != expected_epoch:
+                    raise HistoryEpochChanged(epoch)
+                yield conn
+                conn.commit()
+            except BaseException:
+                conn.rollback()
+                raise
+        finally:
+            conn.execute(f"PRAGMA busy_timeout={timeout}")
+
+    def write_execution_start(self, command: RunStart) -> bool:
+        encode_command(command)
+        if type(command) is not RunStart:
+            raise ValueError("Expected execution start command")
+        with self._execution_transaction(command.storage_epoch) as conn:
+            conn.execute(
+                "INSERT INTO execution_runs "
+                "(run_id,session_id,queue_item_id,job_id,storage_epoch,record_version,"
+                "observed_dequeue_at,outcome,diagnostics_omitted) "
+                "VALUES(?,?,?,?,?,1,?,'observed',0) "
+                "ON CONFLICT(session_id,queue_item_id) DO NOTHING",
+                (
+                    command.run_id,
+                    command.session_id,
+                    command.queue_item_id,
+                    command.job_id,
+                    command.storage_epoch,
+                    command.observed_dequeue_at,
+                ),
+            )
+        return True
+
+    def write_execution_finish(self, command: RunFinish) -> bool:
+        encode_command(command)
+        if type(command) is not RunFinish:
+            raise ValueError("Expected execution finish command")
+        with self._execution_transaction(command.storage_epoch) as conn:
+            row = conn.execute(
+                "SELECT * FROM execution_runs WHERE session_id=? AND queue_item_id=? "
+                "AND storage_epoch=?",
+                (command.session_id, command.queue_item_id, command.storage_epoch),
+            ).fetchone()
+            if row is None:
+                return False
+            completion = classify_evidence(row["job_id"], command.evidence)
+            if row["outcome"] == "unknown":
+                return True
+            if row["outcome"] != "observed":
+                contradictory = completion.reason == "conflicting_terminal" or (
+                    completion.outcome in ("succeeded", "failed", "interrupted")
+                    and completion.outcome != row["outcome"]
+                )
+                if contradictory:
+                    conn.execute(
+                        "UPDATE execution_runs SET outcome='unknown',"
+                        "reason='conflicting_terminal' WHERE run_id=?",
+                        (row["run_id"],),
+                    )
+                return True
+            conn.execute(
+                "UPDATE execution_runs SET observed_callback_at=?,"
+                "executor_started_at=?,"
+                "terminal_event_at=?,outcome=?,reason=?,diagnostics_omitted=?,"
+                "node_id=?,"
+                "node_type=?,exception_type=? WHERE run_id=?",
+                (
+                    command.observed_callback_at,
+                    completion.executor_started_at,
+                    completion.terminal_event_at,
+                    completion.outcome,
+                    completion.reason,
+                    int(completion.diagnostics_omitted),
+                    completion.node_id,
+                    completion.node_type,
+                    completion.exception_type,
+                    row["run_id"],
+                ),
+            )
+        return True
+
+    def get_execution_run(self, run_id: str) -> dict | None:
+        row = (
+            self.model.get_connection()
+            .execute("SELECT * FROM execution_runs WHERE run_id=?", (run_id,))
+            .fetchone()
+        )
+        return dict(row) if row is not None else None

database/execution_history_schema.py

@@ -0,0 +1,349 @@
+"""Versioned additive schema for observed execution attempts."""
+
+import re
+import sqlite3
+from uuid import uuid4
+
+SCHEMA_VERSION = 3
+
+
+def _uuid_check(column: str) -> str:
+    return (
+        f"typeof({column}) = 'text' AND length({column}) = 36 "
+        f"AND substr({column}, 9, 1) = '-' AND substr({column}, 14, 1) = '-' "
+        f"AND substr({column}, 19, 1) = '-' AND substr({column}, 24, 1) = '-' "
+        f"AND length(replace({column}, '-', '')) = 32 "
+        f"AND replace({column}, '-', '') NOT GLOB '*[^0-9a-f]*'"
+    )
+
+
+def _integer_check(column: str, *, nullable: bool = False) -> str:
+    expression = f"typeof({column}) = 'integer' AND {column} >= 0"
+    return f"{column} IS NULL OR ({expression})" if nullable else expression
+
+
+_METADATA_CHECKS = (
+    "singleton_id = 1",
+    "schema_version = 3",
+    _uuid_check("storage_epoch"),
+)
+_RUN_CHECKS = (
+    *(
+        _uuid_check(column)
+        for column in ("run_id", "session_id", "job_id", "storage_epoch")
+    ),
+    _integer_check("queue_item_id"),
+    "record_version = 1",
+    _integer_check("observed_dequeue_at"),
+    *(
+        _integer_check(column, nullable=True)
+        for column in (
+            "observed_callback_at",
+            "executor_started_at",
+            "terminal_event_at",
+        )
+    ),
+    "outcome IN ('observed', 'succeeded', 'failed', 'interrupted', 'unknown')",
+    "(outcome = 'observed' AND reason IS NULL AND "
+    "observed_callback_at IS NULL) OR (outcome != 'observed' AND "
+    "reason IS NOT NULL AND observed_callback_at IS NOT NULL AND "
+    "((outcome = 'succeeded' AND reason = 'terminal_success') OR "
+    "(outcome = 'failed' AND reason = 'terminal_error') OR (outcome = "
+    "'interrupted' AND reason = 'terminal_interrupted') OR (outcome = "
+    "'unknown' AND reason IN ('missing_terminal', "
+    "'conflicting_terminal', 'invalid_evidence', 'evidence_limit'))))",
+    "diagnostics_omitted IN (0, 1)",
+    *(
+        f"{column} IS NULL OR (typeof({column}) = 'text' "
+        f"AND length({column}) <= {limit})"
+        for column, limit in (
+            ("node_id", 128),
+            ("node_type", 128),
+            ("exception_type", 256),
+        )
+    ),
+)
+_TABLE_COLUMNS = {
+    "execution_history_metadata": (
+        ("singleton_id", "INTEGER", 1, 1),
+        ("schema_version", "INTEGER", 1, 0),
+        ("storage_epoch", "TEXT", 1, 0),
+    ),
+    "execution_runs": (
+        ("run_id", "TEXT", 1, 1),
+        ("session_id", "TEXT", 1, 0),
+        ("queue_item_id", "INTEGER", 1, 0),
+        ("job_id", "TEXT", 1, 0),
+        ("storage_epoch", "TEXT", 1, 0),
+        ("record_version", "INTEGER", 1, 0),
+        ("observed_dequeue_at", "INTEGER", 1, 0),
+        ("observed_callback_at", "INTEGER", 0, 0),
+        ("executor_started_at", "INTEGER", 0, 0),
+        ("terminal_event_at", "INTEGER", 0, 0),
+        ("outcome", "TEXT", 1, 0),
+        ("reason", "TEXT", 0, 0),
+        ("diagnostics_omitted", "INTEGER", 1, 0),
+        ("node_id", "TEXT", 0, 0),
+        ("node_type", "TEXT", 0, 0),
+        ("exception_type", "TEXT", 0, 0),
+    ),
+}
+_TABLE_CHECKS = {
+    "execution_history_metadata": _METADATA_CHECKS,
+    "execution_runs": _RUN_CHECKS,
+}
+# Row-only guards preserve the supported v1 DDL fingerprint while bounding reads.
+_ROW_BYTE_CHECKS = {
+    "execution_history_metadata": ("length(CAST(storage_epoch AS BLOB)) = 36",),
+    "execution_runs": (
+        *(
+            f"length(CAST({column} AS BLOB)) = 36"
+            for column in (
+                "run_id",
+                "session_id",
+                "job_id",
+                "storage_epoch",
+            )
+        ),
+        *(
+            f"{column} IS NULL OR length(CAST({column} AS BLOB)) <= {limit}"
+            for column, limit in (
+                ("node_id", 512),
+                ("node_type", 512),
+                ("exception_type", 1024),
+            )
+        ),
+    ),
+}
+
+_INDEXES = {
+    "idx_execution_runs_identity": (True, ("session_id", "queue_item_id")),
+    "idx_execution_runs_observed": (False, ("observed_dequeue_at", "run_id")),
+}
+
+
+def _compact(sql: str) -> str:
+    return "".join(
+        token if token.startswith("'") else re.sub(r"\s+", "", token).lower()
+        for token in re.findall(r"'(?:''|[^'])*'|[^']+", sql)
+    )
+
+
+def history_schema_error(
+    conn: sqlite3.Connection, *, allow_missing: bool = True
+) -> str | None:
+    """Reject partial/unsupported history schemas and invalid persisted records."""
+    tables = {
+        row[0]: row[1]
+        for row in conn.execute("SELECT name,sql FROM sqlite_master WHERE type='table'")
+        if row[0] in _TABLE_COLUMNS
+    }
+    from .execution_context_schema import TABLE_COLUMNS, context_schema_error
+
+    from .execution_output_schema import (
+        TABLE_COLUMNS as OUTPUT_COLUMNS,
+        output_schema_error,
+    )
+
+    output_present = any(
+        row[0] in OUTPUT_COLUMNS
+        for row in conn.execute("SELECT name FROM sqlite_master")
+    )
+    if not tables and output_present:
+        return "Execution outputs require execution history"
+    context_present = any(
+        row[0] in TABLE_COLUMNS
+        for row in conn.execute("SELECT name FROM sqlite_master WHERE type='table'")
+    )
+    if not tables and context_present:
+        return "Execution contexts require execution history"
+    if not tables:
+        return None if allow_missing else "Missing execution history schema"
+    if tables.keys() != _TABLE_COLUMNS.keys():
+        return "Incomplete execution history schema"
+    metadata = conn.execute(
+        "SELECT CASE WHEN typeof(schema_version) = 'integer' "
+        "THEN schema_version ELSE NULL END FROM execution_history_metadata LIMIT 2"
+    ).fetchall()
+    if len(metadata) != 1 or len(metadata[0]) != 1 or metadata[0][0] not in (1, 2, 3):
+        return "Unsupported execution history metadata/version"
+    version = metadata[0][0]
+    if version < 3 and output_present:
+        return "Unexpected execution outputs for history v1/v2"
+    if version == 1 and context_present:
+        return "Unexpected execution contexts for history v1"
+    if not allow_missing and version != SCHEMA_VERSION:
+        return "Execution history migration required"
+    for table, expected in _TABLE_COLUMNS.items():
+        columns = tuple(
+            (row[1], row[2].upper(), row[3], row[5])
+            for row in conn.execute(f"PRAGMA table_info({table})")
+        )
+        if columns != expected:
+            return f"{table} has invalid execution history columns"
+        sql = _compact(tables[table])
+        checks = _TABLE_CHECKS[table]
+        if table == "execution_history_metadata":
+            checks = tuple(
+                check.replace("schema_version = 3", f"schema_version = {version}")
+                for check in checks
+            )
+        if sql.count("check(") != len(checks) or any(
+            f"check({_compact(check)})" not in sql for check in checks
+        ):
+            return f"{table} has invalid execution history checks"
+        if conn.execute(f"PRAGMA foreign_key_list({table})").fetchone():
+            return f"{table} has unexpected execution history foreign keys"
+        for index in conn.execute(f"PRAGMA index_list({table})"):
+            if not index[2]:
+                continue
+            if table == "execution_runs" and index[1] == "idx_execution_runs_identity":
+                continue
+            name = index[1].replace('"', '""')
+            keys = tuple(
+                row for row in conn.execute(f'PRAGMA index_xinfo("{name}")') if row[5]
+            )
+            primary = tuple(column[0] for column in expected if column[3])
+            if (
+                index[3] != "pk"
+                or index[4]
+                or tuple(key[2] for key in keys) != primary
+                or any(key[3] or key[4] != "BINARY" for key in keys)
+            ):
+                return f"{table} has unexpected execution history uniqueness"
+        predicate = " AND ".join(
+            f"({check})" for check in (*checks, *_ROW_BYTE_CHECKS[table])
+        )
+        if conn.execute(
+            f"SELECT 1 FROM {table} WHERE NOT ({predicate}) LIMIT 1"
+        ).fetchone():
+            return f"{table} contains invalid execution history records"
+    metadata = conn.execute(
+        "SELECT singleton_id,schema_version,storage_epoch FROM "
+        "execution_history_metadata"
+    ).fetchall()
+    if len(metadata) != 1 or metadata[0][0] != 1 or metadata[0][1] != version:
+        return "Unsupported execution history metadata/version"
+    indexes = {row[1]: row for row in conn.execute("PRAGMA index_list(execution_runs)")}
+    for name, (unique, columns) in _INDEXES.items():
+        row = indexes.get(name)
+        if row is None or bool(row[2]) != unique or row[4]:
+            return f"Missing or invalid execution history index {name}"
+        actual = tuple(conn.execute(f"PRAGMA index_xinfo({name})"))
+        keys = tuple(item for item in actual if item[5])
+        if tuple(item[2] for item in keys) != columns or any(
+            item[3] or item[4] != "BINARY" for item in keys
+        ):
+            return f"Invalid execution history index {name}"
+    if version >= 2:
+        error = context_schema_error(conn)
+        if error:
+            return error
+    return output_schema_error(conn) if version == 3 else None
+
+
+def install_context_schema(conn: sqlite3.Connection) -> None:
+    from .execution_context_schema import install_context_schema as install
+
+    install(conn)
+
+
+def install_output_schema(conn: sqlite3.Connection) -> None:
+    from .execution_output_schema import install_output_schema as install
+
+    install(conn)
+
+
+def migrate_execution_history(conn: sqlite3.Connection) -> None:
+    """Install history, contexts and outputs atomically, without backfill."""
+    conn.execute("SAVEPOINT execution_history_schema")
+    try:
+        error = history_schema_error(conn)
+        if error:
+            raise sqlite3.DatabaseError(error)
+        present = conn.execute(
+            "SELECT 1 FROM sqlite_master WHERE name IN "
+            "('execution_runs','execution_history_metadata') LIMIT 1"
+        ).fetchone()
+        if not present:
+            for table, columns in _TABLE_COLUMNS.items():
+                definitions = [
+                    f"{name} {kind}{' NOT NULL' if required else ''}"
+                    f"{' PRIMARY KEY' if primary else ''}"
+                    for name, kind, required, primary in columns
+                ]
+                definitions.extend(f"CHECK({check})" for check in _TABLE_CHECKS[table])
+                conn.execute(f"CREATE TABLE {table} ({', '.join(definitions)})")
+            for name, (unique, columns) in _INDEXES.items():
+                conn.execute(
+                    f"CREATE {'UNIQUE ' if unique else ''}INDEX {name} "
+                    f"ON execution_runs({', '.join(columns)})"
+                )
+            conn.execute(
+                "INSERT INTO execution_history_metadata VALUES(1,?,?)",
+                (SCHEMA_VERSION, str(uuid4())),
+            )
+            install_context_schema(conn)
+            install_output_schema(conn)
+        else:
+            version, epoch = conn.execute(
+                "SELECT schema_version,storage_epoch FROM execution_history_metadata"
+            ).fetchone()
+            if version < SCHEMA_VERSION:
+                conn.execute(
+                    "ALTER TABLE execution_history_metadata RENAME TO "
+                    "execution_history_metadata_old"
+                )
+                definitions = [
+                    f"{name} {kind}{' NOT NULL' if required else ''}"
+                    f"{' PRIMARY KEY' if primary else ''}"
+                    for name, kind, required, primary in _TABLE_COLUMNS[
+                        "execution_history_metadata"
+                    ]
+                ]
+                definitions.extend(f"CHECK({check})" for check in _METADATA_CHECKS)
+                conn.execute(
+                    "CREATE TABLE execution_history_metadata "
+                    f"({', '.join(definitions)})"
+                )
+                conn.execute(
+                    "INSERT INTO execution_history_metadata VALUES(1,?,?)",
+                    (SCHEMA_VERSION, epoch),
+                )
+                conn.execute("DROP TABLE execution_history_metadata_old")
+                if version == 1:
+                    install_context_schema(conn)
+                install_output_schema(conn)
+        # Read projections need reverse content lookup and the complete run order.
+        for table, columns in (
+            ("execution_prompt_snapshots", "raw_text_id,snapshot_id"),
+            ("execution_contexts", "positive_snapshot_id,context_id"),
+            ("execution_run_contexts", "context_id,run_id,anchor_id"),
+            ("execution_runs", "observed_dequeue_at,session_id,queue_item_id,run_id"),
+        ):
+            conn.execute(
+                f"CREATE INDEX IF NOT EXISTS idx_history_read_{table} "
+                f"ON {table}({columns})"
+            )
+        error = history_schema_error(conn, allow_missing=False)
+        if error:
+            raise sqlite3.DatabaseError(error)
+        conn.execute("RELEASE execution_history_schema")
+    except BaseException:
+        conn.execute("ROLLBACK TO execution_history_schema")
+        conn.execute("RELEASE execution_history_schema")
+        raise
+
+
+def rotate_history_epoch(conn: sqlite3.Connection) -> str:
+    """Publish a fresh staging epoch without rewriting restored run epochs."""
+    epoch = str(uuid4())
+    with conn:
+        cursor = conn.execute(
+            "UPDATE execution_history_metadata SET storage_epoch=? WHERE "
+            "singleton_id=1",
+            (epoch,),
+        )
+        if cursor.rowcount != 1:
+            raise sqlite3.DatabaseError("Missing execution history metadata")
+    return epoch

database/execution_output_schema.py

@@ -0,0 +1,218 @@
+"""Owned schema and bounded restore validation for frozen output locators."""
+
+import sqlite3
+
+from .execution_history_schema import _compact, _uuid_check
+
+try:
+    from ..utils.execution_outputs import (
+        BUNDLE_OVERHEAD,
+        MAX_BUNDLE_BYTES,
+        MAX_IMAGE_POSITIONS,
+        MAX_REFERENCE_BYTES,
+        REFERENCE_OVERHEAD,
+        STATE_REASONS,
+        OutputBundle,
+        OutputReference,
+        validate_output_bundle,
+    )
+except ImportError:
+    from utils.execution_outputs import (
+        BUNDLE_OVERHEAD,
+        MAX_BUNDLE_BYTES,
+        MAX_IMAGE_POSITIONS,
+        MAX_REFERENCE_BYTES,
+        REFERENCE_OVERHEAD,
+        STATE_REASONS,
+        OutputBundle,
+        OutputReference,
+        validate_output_bundle,
+    )
+
+
+TABLE_COLUMNS = {
+    "execution_output_capture": (
+        ("run_id", "TEXT", 1, 1),
+        ("state", "TEXT", 1, 0),
+        ("reason", "TEXT", 1, 0),
+        ("format_version", "INTEGER", 1, 0),
+        ("source", "TEXT", 1, 0),
+    ),
+    "execution_run_outputs": (
+        ("run_id", "TEXT", 1, 1),
+        ("output_node_id", "TEXT", 1, 2),
+        ("image_ordinal", "INTEGER", 1, 3),
+        ("folder_type", "TEXT", 1, 0),
+        ("subfolder", "TEXT", 1, 0),
+        ("filename", "TEXT", 1, 0),
+        ("path_style", "TEXT", 1, 0),
+        ("root_key", "TEXT", 0, 0),
+    ),
+}
+TABLE_FOREIGN_KEYS = {
+    "execution_output_capture": (("run_id", "execution_runs", "run_id"),),
+    "execution_run_outputs": (("run_id", "execution_output_capture", "run_id"),),
+}
+
+
+def _text_check(column: str, maximum: int, *, characters: int | None = None) -> str:
+    return (
+        f"typeof({column}) = 'text' AND length({column}) <= "
+        f"{maximum if characters is None else characters} "
+        f"AND length(CAST({column} AS BLOB)) <= {maximum} "
+        f"AND instr(CAST({column} AS BLOB), x'00') = 0"
+    )
+
+
+def _enum_check(column: str, values: tuple | frozenset) -> str:
+    return (
+        f"{_text_check(column, 32)} AND {column} IN ("
+        + ",".join("'" + value + "'" for value in sorted(values))
+        + ")"
+    )
+
+
+_RUN_ID_CHECK = f"{_uuid_check('run_id')} AND {_text_check('run_id', 36)}"
+TABLE_CHECKS = {
+    "execution_output_capture": (
+        _RUN_ID_CHECK,
+        _enum_check("state", tuple(STATE_REASONS)),
+        _enum_check("reason", frozenset.union(*STATE_REASONS.values())),
+        "typeof(format_version) = 'integer' AND format_version = 1",
+        f"{_text_check('source', 18)} AND source = 'host_final_history'",
+    ),
+    "execution_run_outputs": (
+        _RUN_ID_CHECK,
+        f"{_text_check('output_node_id', 512, characters=128)} "
+        "AND length(output_node_id) >= 1",
+        "typeof(image_ordinal) = 'integer' AND image_ordinal BETWEEN 0 AND 255",
+        _enum_check("folder_type", ("output", "temp")),
+        _text_check("subfolder", 2048),
+        f"{_text_check('filename', 1024)} AND length(filename) >= 1",
+        _enum_check("path_style", ("posix", "windows")),
+        f"root_key IS NULL OR ({_text_check('root_key', 64)} "
+        "AND length(root_key) = 64 AND root_key NOT GLOB '*[^0-9a-f]*')",
+    ),
+}
+
+
+def _table_sql(table: str) -> str:
+    columns = TABLE_COLUMNS[table]
+    definitions = [
+        f"{name} {kind}{' NOT NULL' if required else ''}"
+        for name, kind, required, _ in columns
+    ]
+    primary = [name for name, _, _, order in columns if order]
+    definitions.append(f"PRIMARY KEY ({', '.join(primary)})")
+    definitions.extend(f"CHECK({check})" for check in TABLE_CHECKS[table])
+    definitions.extend(
+        f"FOREIGN KEY ({column}) REFERENCES {target}({key}) "
+        "ON UPDATE NO ACTION ON DELETE NO ACTION"
+        for column, target, key in TABLE_FOREIGN_KEYS[table]
+    )
+    return f"CREATE TABLE {table} ({', '.join(definitions)})"
+
+
+def install_output_schema(conn: sqlite3.Connection) -> None:
+    """Install the two owned tables within the caller's migration transaction."""
+    for table in TABLE_COLUMNS:
+        conn.execute(_table_sql(table))
+
+
+def output_schema_error(conn: sqlite3.Connection) -> str | None:
+    """Reject altered schemas and invalid rows before returning bounded DTOs."""
+    if conn.execute(
+        "SELECT 1 FROM sqlite_master WHERE type IN ('trigger','view') LIMIT 1"
+    ).fetchone():
+        return "Unsupported execution output trigger/view"
+    tables = {
+        row[0]: row[1]
+        for row in conn.execute("SELECT name,sql FROM sqlite_master WHERE type='table'")
+        if row[0] in TABLE_COLUMNS
+    }
+    if tables.keys() != TABLE_COLUMNS.keys():
+        return "Incomplete execution output schema"
+    for table, expected in TABLE_COLUMNS.items():
+        columns = tuple(
+            (row[1], row[2].upper(), row[3], row[5])
+            for row in conn.execute(f"PRAGMA table_xinfo({table})")
+        )
+        if columns != expected or _compact(tables[table]) != _compact(
+            _table_sql(table)
+        ):
+            return f"{table} has invalid execution output schema"
+        foreign = tuple(
+            sorted(
+                (row[3], row[2], row[4], row[5], row[6], row[7])
+                for row in conn.execute(f"PRAGMA foreign_key_list({table})")
+            )
+        )
+        if foreign != tuple(
+            (*key, "NO ACTION", "NO ACTION", "NONE")
+            for key in TABLE_FOREIGN_KEYS[table]
+        ):
+            return f"{table} has invalid execution output foreign keys"
+        for index in conn.execute(f"PRAGMA index_list({table})"):
+            name = index[1].replace('"', '""')
+            keys = tuple(
+                row for row in conn.execute(f'PRAGMA index_xinfo("{name}")') if row[5]
+            )
+            if (
+                not index[2]
+                or index[3] != "pk"
+                or index[4]
+                or tuple(key[2] for key in keys)
+                != tuple(column[0] for column in expected if column[3])
+                or any(key[3] or key[4] != "BINARY" for key in keys)
+            ):
+                return f"{table} has unexpected execution output index/uniqueness"
+        # Run every scalar guard in SQLite before any evidence text is fetched.
+        predicate = " AND ".join(f"({check})" for check in TABLE_CHECKS[table])
+        if conn.execute(
+            f"SELECT 1 FROM {table} WHERE NOT ({predicate}) LIMIT 1"
+        ).fetchone():
+            return f"{table} contains invalid execution output records"
+        if conn.execute(f"PRAGMA foreign_key_check({table})").fetchone():
+            return f"{table} contains orphan execution output records"
+    reference_bytes = f"{REFERENCE_OVERHEAD} + " + " + ".join(
+        f"coalesce(length(CAST({column} AS BLOB)), 0)"
+        for column in (
+            "output_node_id",
+            "folder_type",
+            "subfolder",
+            "filename",
+            "path_style",
+            "root_key",
+        )
+    )
+    if conn.execute(
+        "SELECT 1 FROM execution_run_outputs "
+        f"WHERE ({reference_bytes}) > {MAX_REFERENCE_BYTES} LIMIT 1"
+    ).fetchone():
+        return "Oversized execution output reference"
+    if conn.execute(
+        "SELECT 1 FROM execution_run_outputs GROUP BY run_id "
+        f"HAVING COUNT(*) > {MAX_IMAGE_POSITIONS} OR "
+        f"{BUNDLE_OVERHEAD} + SUM({reference_bytes}) > {MAX_BUNDLE_BYTES} LIMIT 1"
+    ).fetchone():
+        return "Oversized execution output bundle"
+    try:
+        for run_id, state, reason, version, source in conn.execute(
+            "SELECT run_id,state,reason,format_version,source "
+            "FROM execution_output_capture"
+        ):
+            references = tuple(
+                OutputReference(*row)
+                for row in conn.execute(
+                    "SELECT output_node_id,image_ordinal,folder_type,subfolder,"
+                    "filename,path_style,root_key FROM execution_run_outputs "
+                    "WHERE run_id=? ORDER BY output_node_id,image_ordinal",
+                    (run_id,),
+                )
+            )
+            validate_output_bundle(
+                OutputBundle(references, state, reason, version, source)
+            )
+    except (ValueError, TypeError, OverflowError, sqlite3.Error) as exc:
+        return f"Invalid execution output content: {type(exc).__name__}"
+    return None

database/execution_outputs.py

@@ -0,0 +1,153 @@
+"""Immutable run-owned output references, independent of mutable galleries."""
+
+import sqlite3
+from contextlib import ExitStack
+from typing import Callable, ContextManager
+
+try:
+    from ..utils.execution_history_events import RunFinish, encode_command
+    from ..utils.execution_outputs import OutputBundle, validate_output_bundle
+except ImportError:
+    from utils.execution_history_events import RunFinish, encode_command
+    from utils.execution_outputs import OutputBundle, validate_output_bundle
+
+
+_REFERENCE_FIELDS = (
+    "output_node_id",
+    "image_ordinal",
+    "folder_type",
+    "subfolder",
+    "filename",
+    "path_style",
+    "root_key",
+)
+_REFERENCE_COLUMNS = ",".join(_REFERENCE_FIELDS)
+
+
+def _capture(conn: sqlite3.Connection, run_id: str) -> sqlite3.Row | None:
+    capture = conn.execute(
+        "SELECT state,reason,format_version,source FROM execution_output_capture "
+        "WHERE run_id=?",
+        (run_id,),
+    ).fetchone()
+    # Legacy callers may disable FK enforcement; never adopt orphan evidence.
+    if capture is None:
+        if conn.execute(
+            "SELECT 1 FROM execution_run_outputs WHERE run_id=? LIMIT 1", (run_id,)
+        ).fetchone():
+            raise sqlite3.DatabaseError("Output references have no capture header")
+    elif (
+        conn.execute(
+            "SELECT 1 FROM execution_runs WHERE run_id=?", (run_id,)
+        ).fetchone()
+        is None
+    ):
+        raise sqlite3.DatabaseError("Output capture has no execution parent")
+    return capture
+
+
+def _references(conn: sqlite3.Connection, run_id: str) -> list[tuple]:
+    return [
+        tuple(row)
+        for row in conn.execute(
+            f"SELECT {_REFERENCE_COLUMNS} FROM execution_run_outputs "
+            "WHERE run_id=? ORDER BY output_node_id,image_ordinal",
+            (run_id,),
+        )
+    ]
+
+
+class ExecutionOutputOperations:
+    """Compose with ExecutionHistoryOperations on PromptDatabase."""
+
+    def write_execution_outputs_for_finish(
+        self,
+        command: RunFinish,
+        bundle: OutputBundle,
+        *,
+        generation_guard: Callable[[], ContextManager[bool]] | None = None,
+    ) -> bool:
+        encode_command(command)
+        if type(command) is not RunFinish:
+            raise ValueError("Expected execution finish command")
+        validate_output_bundle(bundle)
+        header = (bundle.state, bundle.reason, bundle.format_version, bundle.source)
+        prepared = [
+            tuple(getattr(reference, field) for field in _REFERENCE_FIELDS)
+            for reference in bundle.references
+        ]
+        # Enter after BEGIN and retain the guard until COMMIT/ROLLBACK finishes.
+        with ExitStack() as guards:
+            with self._execution_transaction(command.storage_epoch) as conn:
+                if generation_guard is not None and not guards.enter_context(
+                    generation_guard()
+                ):
+                    return False
+                parent = conn.execute(
+                    "SELECT run_id FROM execution_runs WHERE session_id=? "
+                    "AND queue_item_id=? AND storage_epoch=?",
+                    (command.session_id, command.queue_item_id, command.storage_epoch),
+                ).fetchone()
+                if parent is None:
+                    return False
+                run_id = parent[0]
+                capture = _capture(conn, run_id)
+                if capture is not None:
+                    if capture[0] != "conflicting" and (
+                        tuple(capture) != header
+                        or _references(conn, run_id) != prepared
+                    ):
+                        conn.execute(
+                            "UPDATE execution_output_capture SET "
+                            "state='conflicting',reason='conflicting' WHERE run_id=?",
+                            (run_id,),
+                        )
+                    return True
+                conn.execute(
+                    "INSERT INTO execution_output_capture "
+                    "(run_id,state,reason,format_version,source) VALUES(?,?,?,?,?)",
+                    (run_id, *header),
+                )
+                conn.executemany(
+                    f"INSERT INTO execution_run_outputs (run_id,{_REFERENCE_COLUMNS}) "
+                    "VALUES(?,?,?,?,?,?,?,?)",
+                    [(run_id, *reference) for reference in prepared],
+                )
+        return True
+
+    def get_execution_outputs(self, run_id: str) -> dict:
+        """Read frozen evidence in one snapshot without resolving its locators."""
+        conn = self.model.get_connection()
+        own_transaction = not conn.in_transaction
+        if own_transaction:
+            conn.execute("BEGIN")
+        try:
+            capture = _capture(conn, run_id)
+            if capture is None:
+                return {
+                    "state": "not_captured",
+                    "reason": None,
+                    "format_version": None,
+                    "source": None,
+                    "references": [],
+                }
+            references = []
+            for row in _references(conn, run_id):
+                reference = dict(zip(_REFERENCE_FIELDS, row))
+                reference.update(
+                    file_identity="unverified",
+                    newly_generated="unknown",
+                    context_membership="unknown",
+                    transient=reference["folder_type"] == "temp",
+                )
+                references.append(reference)
+            return {
+                "state": capture[0],
+                "reason": capture[1],
+                "format_version": capture[2],
+                "source": capture[3],
+                "references": references,
+            }
+        finally:
+            if own_transaction:
+                conn.rollback()

database/history_gallery.py

@@ -0,0 +1,140 @@
+"""Bounded exact-path saved display within the caller's history snapshot."""
+
+import os
+import sqlite3
+
+try:
+    from ..utils.inspection_response import (
+        MAX_LIBRARY_BYTES,
+        _library_projection,
+        library_metadata,
+    )
+except ImportError:
+    from utils.inspection_response import (
+        MAX_LIBRARY_BYTES,
+        _library_projection,
+        library_metadata,
+    )
+
+
+def _display_bytes(value: bytes | None) -> str | None:
+    if value is None:
+        return None
+    return value.decode("utf-8", "replace")
+
+
+def _tags(
+    conn: sqlite3.Connection, owner: int, *, pair: bool
+) -> tuple[list[str], bool]:
+    table, column = (
+        ("prompt_pair_tags", "pair_id") if pair else ("prompt_tags", "prompt_id")
+    )
+    tags = []
+    used = 0
+    cursor = conn.execute(
+        f"SELECT substr(CAST(t.name AS BLOB),1,?) FROM {table} pt "
+        f"JOIN tags t ON t.id=pt.tag_id WHERE pt.{column}=? LIMIT 129",
+        (MAX_LIBRARY_BYTES + 1, owner),
+    )
+    for row in cursor:
+        used += len(row[0])
+        if used > MAX_LIBRARY_BYTES or len(tags) == 128:
+            return [], True
+        tags.append(_display_bytes(row[0]))
+    return tags, False
+
+
+def _saved_projection(conn: sqlite3.Connection, row: sqlite3.Row) -> tuple[dict, dict]:
+    saved = {"prompt_id": row["prompt_id"]}
+    pairs = list(
+        conn.execute(
+            "SELECT pp.id,pp.positive_prompt_id,"
+            "substr(CAST(pp.title AS BLOB),1,?) AS title,"
+            "substr(CAST(pp.category AS BLOB),1,?) AS category "
+            "FROM image_prompt_pairs ip JOIN prompt_pairs pp ON pp.id=ip.pair_id "
+            "WHERE ip.image_id=? LIMIT 2",
+            (MAX_LIBRARY_BYTES + 1, MAX_LIBRARY_BYTES + 1, row["id"]),
+        )
+    )
+    if len(pairs) > 1:
+        return saved, _library_projection({"status": "limited"})
+    pair = pairs[0] if pairs else None
+    title = pair["title"] if pair is not None else None
+    recipe_tags, limited = (
+        _tags(conn, pair["id"], pair=True) if pair is not None else ([], False)
+    )
+    positive_tags = []
+    if not recipe_tags and not limited:
+        # D39: only read the positive's tags when they are what the panel shows.
+        positive_tags, limited = _tags(conn, row["prompt_id"], pair=False)
+    metadata = library_metadata(
+        row["category"],
+        positive_tags,
+        (pair["category"], recipe_tags) if pair is not None else None,
+    )
+    limited = limited or any(
+        value is not None and len(value) > MAX_LIBRARY_BYTES
+        for value in (title, metadata["category"])
+    )
+    if pair is not None:
+        saved["pair"] = {
+            "id": pair["id"],
+            "positive_prompt_id": pair["positive_prompt_id"],
+        }
+    library = _library_projection(
+        {
+            "status": "linked",
+            "positive_id": row["prompt_id"],
+            "pair_id": pair["id"] if pair is not None else None,
+            "title": _display_bytes(title),
+            **metadata,
+            "category": _display_bytes(metadata["category"]),
+        }
+    )
+    if limited:
+        library = _library_projection({"status": "limited", "tags": []})
+    return saved, library
+
+
+def read_history_gallery(
+    conn: sqlite3.Connection, paths: tuple[str, ...]
+) -> dict[str, dict]:
+    """Read at most two rows/path and two pair associations/unique row; no writes."""
+    if not conn.in_transaction:
+        raise sqlite3.ProgrammingError("Saved history display requires caller snapshot")
+    if (
+        type(paths) is not tuple
+        or len(paths) > 100
+        or any(
+            type(path) is not str
+            or not os.path.isabs(path)
+            or path != os.path.normcase(os.path.normpath(path))
+            or "\0" in path
+            for path in paths
+        )
+    ):
+        raise ValueError("Invalid history gallery paths")
+    result = {}
+    for path in dict.fromkeys(paths):
+        rows = list(
+            conn.execute(
+                "SELECT gi.id,gi.prompt_id,"
+                "substr(CAST(p.category AS BLOB),1,?) AS category "
+                "FROM generated_images gi JOIN prompts p ON p.id=gi.prompt_id "
+                "WHERE gi.file_path=? OR gi.image_path=? LIMIT 2",
+                (MAX_LIBRARY_BYTES + 1, path, path),
+            )
+        )
+        match = "none" if not rows else "unique" if len(rows) == 1 else "ambiguous"
+        saved, library = None, _library_projection(
+            {"status": "none" if not rows else "ambiguous"}
+        )
+        if match == "unique":
+            saved, library = _saved_projection(conn, rows[0])
+        result[path] = {
+            "gallery_match": match,
+            "gallery_image_id": rows[0]["id"] if match == "unique" else None,
+            "saved": saved,
+            "library": library,
+        }
+    return result

database/history_reads.py

@@ -0,0 +1,585 @@
+"""Snapshot-owned, bounded projections of immutable execution evidence."""
+
+import codecs
+from contextlib import contextmanager
+from contextvars import ContextVar
+from dataclasses import dataclass
+import hashlib
+import json
+import re
+import sqlite3
+from typing import Iterator
+
+from .execution_contexts import _hints, _values, content_digest
+from .execution_history import HistoryEpochChanged
+from .execution_outputs import _capture, _REFERENCE_COLUMNS
+from .library import _ELIGIBLE
+
+RunKey = tuple[int, str, int, str]
+
+
+class HistoryCaptureChanged(ValueError):
+    """A continuation no longer describes the stored output capture header."""
+
+
+@dataclass(frozen=True)
+class WindowPosition:
+    raw_text_id: str
+    upper: RunKey
+    before: RunKey
+
+
+@dataclass(frozen=True)
+class OutputPosition:
+    capture_fingerprint: str
+    after: tuple[str, int]
+
+
+_DIGEST = re.compile(r"[0-9a-f]{64}")
+_UUID = re.compile(r"[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12}")
+
+
+def _valid_digest(value: object) -> bool:
+    return type(value) is str and _DIGEST.fullmatch(value) is not None
+
+
+def _valid_run_key(value: object) -> bool:
+    return (
+        type(value) is tuple
+        and len(value) == 4
+        and all(type(value[i]) is int and 0 <= value[i] <= 2**63 - 1 for i in (0, 2))
+        and all(
+            type(value[i]) is str and _UUID.fullmatch(value[i]) is not None
+            for i in (1, 3)
+        )
+    )
+
+
+def _valid_coordinate(value: object) -> bool:
+    if type(value) is not tuple or len(value) != 2:
+        return False
+    node, ordinal = value
+    if not (
+        type(node) is str
+        and 1 <= len(node) <= 128
+        and "\0" not in node
+        and type(ordinal) is int
+        and 0 <= ordinal <= 255
+    ):
+        return False
+    try:
+        return len(node.encode("utf-8")) <= 512
+    except UnicodeError:
+        return False
+
+
+def _validate_window_position(position: WindowPosition | None) -> None:
+    if position is not None and not (
+        type(position) is WindowPosition
+        and _valid_digest(position.raw_text_id)
+        and _valid_run_key(position.upper)
+        and _valid_run_key(position.before)
+    ):
+        raise ValueError("Invalid history window position")
+
+
+def _validate_output_position(position: OutputPosition | None) -> None:
+    if position is not None and not (
+        type(position) is OutputPosition
+        and _valid_digest(position.capture_fingerprint)
+        and _valid_coordinate(position.after)
+    ):
+        raise ValueError("Invalid history output position")
+
+
+_snapshots: ContextVar[tuple] = ContextVar("history_read_snapshots", default=())
+_KEY = "(r.observed_dequeue_at,r.session_id,r.queue_item_id,r.run_id)"
+_ORDER = (
+    "r.observed_dequeue_at DESC,r.session_id COLLATE BINARY DESC,"
+    "r.queue_item_id DESC,r.run_id COLLATE BINARY DESC"
+)
+# Stale-hint rows a newest-pair lookup inspects before it reports none.
+_PAIR_RUN_CANDIDATES = 16
+_MATCHING = (
+    "SELECT DISTINCT rc.run_id FROM execution_prompt_snapshots s "
+    "JOIN execution_contexts c ON c.positive_snapshot_id=s.snapshot_id "
+    "JOIN execution_run_contexts rc ON rc.context_id=c.context_id "
+    "WHERE s.raw_text_id=?"
+)
+
+
+def _limit(limit: int) -> None:
+    if type(limit) is not int or not 1 <= limit <= 100:
+        raise ValueError("History limit must be an integer from 1 to 100")
+
+
+def _key(row: dict) -> RunKey:
+    return tuple(
+        row[name]
+        for name in ("observed_dequeue_at", "session_id", "queue_item_id", "run_id")
+    )
+
+
+def _context_capture(conn: sqlite3.Connection, run_id: str) -> dict:
+    row = conn.execute(
+        "SELECT state,reason,format_version FROM execution_context_capture "
+        "WHERE run_id=?",
+        (run_id,),
+    ).fetchone()
+    return (
+        dict(row)
+        if row is not None
+        else {"state": "not_captured", "reason": "missing", "format_version": None}
+    )
+
+
+def _output_header(conn: sqlite3.Connection, run_id: str) -> tuple[dict, int | None]:
+    row = _capture(conn, run_id)
+    if row is None:
+        return (
+            dict(state="not_captured", reason=None, format_version=None, source=None),
+            None,
+        )
+    count = conn.execute(
+        "SELECT COUNT(*) FROM execution_run_outputs WHERE run_id=?", (run_id,)
+    ).fetchone()[0]
+    return dict(row), count
+
+
+def _list_status(capture: dict) -> str:
+    if capture["state"] == "not_captured":
+        return "not_captured"
+    return "complete" if capture["state"] == "complete" else "incomplete"
+
+
+def _output_summary(conn: sqlite3.Connection, run_id: str) -> dict:
+    capture, count = _output_header(conn, run_id)
+    return dict(
+        capture, retained_occurrence_count=count, list_status=_list_status(capture)
+    )
+
+
+def _fingerprint(capture: dict, count: int | None) -> str:
+    return hashlib.sha256(
+        json.dumps(
+            [
+                1,
+                *(capture[k] for k in ("state", "reason", "format_version", "source")),
+                count,
+            ],
+            ensure_ascii=False,
+            allow_nan=False,
+            separators=(",", ":"),
+        ).encode("utf-8")
+    ).hexdigest()
+
+
+def _receipt(run_id: str, row: sqlite3.Row) -> dict:
+    return {
+        "occurrence": {
+            "run_id": run_id,
+            "output_node_id": row["output_node_id"],
+            "image_ordinal": row["image_ordinal"],
+        },
+        "reference": {
+            key: row[key]
+            for key in (
+                "folder_type",
+                "subfolder",
+                "filename",
+                "path_style",
+                "root_key",
+            )
+        },
+        "file_identity": "unverified",
+        "newly_generated": "unknown",
+        "context_membership": "unknown",
+        "transient": row["folder_type"] == "temp",
+    }
+
+
+class HistoryReadOperations:
+    """Read evidence only inside an explicitly owned, epoch-first snapshot."""
+
+    @contextmanager
+    def recorded_history_snapshot(
+        self, expected_epoch: str | None = None
+    ) -> Iterator[str]:
+        conn = self.model.get_connection()
+        if conn.in_transaction:
+            raise sqlite3.ProgrammingError(
+                "History reads require their own transaction"
+            )
+        conn.execute("BEGIN")
+        token = _snapshots.set((*_snapshots.get(), conn))
+        try:
+            epoch = self.get_execution_epoch()
+            if expected_epoch is not None and epoch != expected_epoch:
+                raise HistoryEpochChanged(epoch)
+            yield epoch
+        finally:
+            try:
+                conn.rollback()
+            finally:
+                _snapshots.reset(token)
+
+    def _history_connection(self) -> sqlite3.Connection:
+        conn = self.model.get_connection()
+        if not conn.in_transaction or conn not in _snapshots.get():
+            raise sqlite3.ProgrammingError(
+                "History read requires recorded_history_snapshot"
+            )
+        return conn
+
+    def list_history_window(
+        self,
+        positive_id: int,
+        *,
+        limit: int = 40,
+        after: WindowPosition | None = None,
+        restart: bool = False,
+    ) -> dict | None:
+        conn = self._history_connection()
+        _limit(limit)
+        _validate_window_position(after)
+        if type(restart) is not bool or (
+            restart and (after is None or after.before != after.upper)
+        ):
+            raise ValueError("Invalid history restart position")
+        saved = conn.execute(
+            "SELECT CASE WHEN typeof(p.text)='text' AND "
+            "length(CAST(p.text AS BLOB))<=65536 THEN p.text ELSE NULL END AS text "
+            f"FROM prompts p WHERE p.id=? AND ({_ELIGIBLE})",
+            (positive_id,),
+        ).fetchone()
+        if saved is None:
+            return None
+        text = saved["text"]
+        raw_id = None if text is None else content_digest("text", (text,))
+        if after is not None and after.raw_text_id != raw_id:
+            raise ValueError("History position text scope changed")
+        group = {
+            "positive_id": positive_id,
+            "raw_text_id": raw_id,
+            "relation": "exact_frozen_raw_text",
+            "scope_status": "uncapturable_text" if text is None else "no_matches",
+        }
+        result = dict(
+            group=group,
+            recorded_run_count=0,
+            items=[],
+            context_summaries=[],
+            has_more=False,
+            next_position=None,
+            resume_position=None,
+        )
+        if (
+            raw_id is None
+            or conn.execute(
+                "SELECT 1 FROM execution_context_texts WHERE text_id=? "
+                "AND text_value=? COLLATE BINARY",
+                (raw_id, text),
+            ).fetchone()
+            is None
+        ):
+            return result
+        where = f"r.run_id IN ({_MATCHING})"
+        params = [raw_id]
+        if after is not None:
+            where += f" AND {_KEY}<=(?,?,?,?)"
+            params.extend(after.upper)
+        count = conn.execute(
+            f"SELECT COUNT(*) FROM execution_runs r WHERE {where}", params
+        ).fetchone()[0]
+        result["recorded_run_count"] = count
+        group["scope_status"] = "matched" if count else "no_matches"
+        if after is not None and not restart:
+            where += f" AND {_KEY}<(?,?,?,?)"
+            params.extend(after.before)
+        rows = [
+            dict(row)
+            for row in conn.execute(
+                f"SELECT r.* FROM execution_runs r WHERE {where} "
+                f"ORDER BY {_ORDER} LIMIT ?",
+                (*params, limit + 1),
+            )
+        ]
+        more = len(rows) > limit
+        rows = rows[:limit]
+        if not rows:
+            return result
+        upper = after.upper if after is not None else _key(rows[0])
+        result["resume_position"] = WindowPosition(raw_id, upper, upper)
+        if more:
+            result["next_position"] = WindowPosition(raw_id, upper, _key(rows[-1]))
+        result["has_more"] = more
+        rows.reverse()
+        by_run = {row["run_id"]: row for row in rows}
+        for row in rows:
+            row["diagnostics_available"] = any(
+                row[field] is not None
+                for field in ("node_id", "node_type", "exception_type")
+            )
+            for field in ("node_id", "node_type", "exception_type"):
+                del row[field]
+            row["diagnostics_omitted"] = bool(row["diagnostics_omitted"])
+            row["context_capture"] = _context_capture(conn, row["run_id"])
+            row["output_summary"] = _output_summary(conn, row["run_id"])
+            row["matching_anchors"] = []
+        placeholders = ",".join("?" for _ in rows)
+        anchors = conn.execute(
+            "SELECT rc.run_id,rc.anchor_id,rc.context_id,rc.evidence_json,"
+            "c.positive_snapshot_id,c.negative_snapshot_id,"
+            "substr(CAST(t.text_value AS BLOB),1,512) AS prefix,"
+            "length(CAST(t.text_value AS BLOB)) AS byte_length "
+            "FROM execution_run_contexts rc JOIN execution_contexts c "
+            "ON c.context_id=rc.context_id JOIN execution_prompt_snapshots p "
+            "ON p.snapshot_id=c.positive_snapshot_id "
+            "LEFT JOIN execution_prompt_snapshots n "
+            "ON n.snapshot_id=c.negative_snapshot_id "
+            "LEFT JOIN execution_context_texts t ON t.text_id=n.raw_text_id "
+            f"WHERE rc.run_id IN ({placeholders}) AND p.raw_text_id=? "
+            "ORDER BY rc.anchor_id COLLATE BINARY,rc.run_id COLLATE BINARY",
+            (*by_run, raw_id),
+        )
+        summaries = {}
+        evidence_by_context = {}
+        for anchor in anchors:
+            reference = {
+                key: anchor[key]
+                for key in (
+                    "anchor_id",
+                    "context_id",
+                    "positive_snapshot_id",
+                    "negative_snapshot_id",
+                )
+            }
+            by_run[anchor["run_id"]]["matching_anchors"].append(reference)
+            context_id = anchor["context_id"]
+            if context_id not in summaries:
+                prefix = b"" if anchor["byte_length"] == 0 else anchor["prefix"]
+                preview = (
+                    None
+                    if prefix is None
+                    else codecs.getincrementaldecoder("utf-8")().decode(
+                        prefix, final=False
+                    )[:100]
+                )
+                evidence_by_context[context_id] = []
+                summaries[context_id] = {
+                    key: anchor[key]
+                    for key in (
+                        "context_id",
+                        "positive_snapshot_id",
+                        "negative_snapshot_id",
+                    )
+                }
+                summaries[context_id]["negative_raw"] = {
+                    "state": (
+                        "unknown"
+                        if preview is None
+                        else "explicit_empty" if not preview else "known"
+                    ),
+                    "preview": preview,
+                    "truncated": (
+                        False
+                        if preview is None
+                        else len(preview.encode("utf-8")) < anchor["byte_length"]
+                    ),
+                    "evidence": evidence_by_context[context_id],
+                }
+            negative = json.loads(anchor["evidence_json"])["negative"]
+            evidence = (
+                negative["raw"]
+                if negative is not None
+                else dict(provenance="unknown", source="unknown", reason="missing")
+            )
+            evidence_by_context[context_id].append(
+                {
+                    "run_id": anchor["run_id"],
+                    "anchor_id": anchor["anchor_id"],
+                    **evidence,
+                }
+            )
+        # Context order follows its first chronological matching run, then anchor.
+        ordered = {}
+        for row in rows:
+            for anchor in row["matching_anchors"]:
+                ordered.setdefault(
+                    anchor["context_id"], summaries[anchor["context_id"]]
+                )
+        result["items"] = rows
+        result["context_summaries"] = list(ordered.values())
+        return result
+
+    def get_newest_pair_run(self, positive_id: int, pair_id: int) -> dict | None:
+        """The newest matching run whose current saved-pair hint names `pair_id`.
+
+        None when the positive is not an eligible saved prompt or the pair is
+        not one of its pairs; ``{"run": None}`` when no recorded run names it.
+        Only a hint captured in this epoch and still resolving to the same pair
+        counts, as in ``get_history_run``.
+        """
+        conn = self._history_connection()
+        saved = conn.execute(
+            "SELECT CASE WHEN typeof(p.text)='text' AND "
+            "length(CAST(p.text AS BLOB))<=65536 THEN p.text ELSE NULL END AS text "
+            f"FROM prompts p WHERE p.id=? AND ({_ELIGIBLE})",
+            (positive_id,),
+        ).fetchone()
+        if saved is None or (
+            conn.execute(
+                "SELECT 1 FROM prompt_pairs WHERE id=? AND positive_prompt_id=?",
+                (pair_id, positive_id),
+            ).fetchone()
+            is None
+        ):
+            return None
+        if saved["text"] is None:
+            return {"run": None}
+        epoch = self.get_execution_epoch()
+        candidates = conn.execute(
+            "SELECT r.run_id,rc.anchor_id,rc.context_id,"
+            "c.positive_snapshot_id,c.negative_snapshot_id "
+            "FROM execution_runs r JOIN execution_run_contexts rc "
+            "ON rc.run_id=r.run_id JOIN execution_contexts c "
+            "ON c.context_id=rc.context_id JOIN execution_prompt_snapshots s "
+            "ON s.snapshot_id=c.positive_snapshot_id "
+            "WHERE s.raw_text_id=? AND rc.pair_id=? AND rc.reference_epoch=? "
+            f"ORDER BY {_ORDER},rc.anchor_id COLLATE BINARY LIMIT ?",
+            (
+                content_digest("text", (saved["text"],)),
+                pair_id,
+                epoch,
+                _PAIR_RUN_CANDIDATES,
+            ),
+        ).fetchall()
+        for row in candidates:
+            positive = _values(conn, row["positive_snapshot_id"])
+            negative = _values(conn, row["negative_snapshot_id"])
+            if positive is None or positive["raw"] != saved["text"]:
+                continue
+            hints = _hints(conn, positive["raw"], negative["raw"] if negative else None)
+            if hints[2] == pair_id:
+                return {
+                    "run": {
+                        key: row[key] for key in ("run_id", "anchor_id", "context_id")
+                    }
+                }
+        return {"run": None}
+
+    def get_history_run(self, run_id: str) -> dict | None:
+        conn = self._history_connection()
+        run = self.get_execution_run(run_id)
+        if run is None:
+            return None
+        capture = self.get_execution_contexts(run_id)
+        snapshots = {
+            row["anchor_id"]: row
+            for row in conn.execute(
+                "SELECT rc.anchor_id,c.positive_snapshot_id,c.negative_snapshot_id,"
+                "p.raw_text_id AS positive_raw_text_id "
+                "FROM execution_run_contexts rc JOIN execution_contexts c "
+                "ON c.context_id=rc.context_id LEFT JOIN execution_prompt_snapshots p "
+                "ON p.snapshot_id=c.positive_snapshot_id WHERE rc.run_id=?",
+                (run_id,),
+            )
+        }
+        contexts = capture.pop("contexts")
+        for context in contexts:
+            row = snapshots[context["anchor_id"]]
+            context.update(
+                {
+                    key: row[key]
+                    for key in (
+                        "positive_snapshot_id",
+                        "negative_snapshot_id",
+                        "positive_raw_text_id",
+                    )
+                }
+            )
+            pair = context["pair_id"]
+            context["navigation"] = {
+                "positive_available": context["positive_prompt_id"] is not None
+                and conn.execute(
+                    f"SELECT 1 FROM prompts p WHERE p.id=? AND ({_ELIGIBLE})",
+                    (context["positive_prompt_id"],),
+                ).fetchone()
+                is not None,
+                "negative_available": context["negative_prompt_id"] is not None,
+                "pair_available": pair is not None
+                and conn.execute(
+                    "SELECT 1 FROM prompt_pairs WHERE id=? AND archived=0", (pair,)
+                ).fetchone()
+                is not None,
+            }
+        return dict(
+            run=run,
+            context_capture=capture,
+            contexts=contexts,
+            output_summary=_output_summary(conn, run_id),
+        )
+
+    def list_history_outputs(
+        self,
+        run_id: str,
+        *,
+        limit: int = 40,
+        after: OutputPosition | None = None,
+    ) -> dict | None:
+        conn = self._history_connection()
+        _limit(limit)
+        _validate_output_position(after)
+        # Validate header/orphans even when the requested parent does not exist.
+        capture, count = _output_header(conn, run_id)
+        if self.get_execution_run(run_id) is None:
+            return None
+        fingerprint = _fingerprint(capture, count)
+        if after is not None and after.capture_fingerprint != fingerprint:
+            raise HistoryCaptureChanged("History output capture changed")
+        where = "run_id=?"
+        params = [run_id]
+        if after is not None:
+            where += " AND (output_node_id COLLATE BINARY,image_ordinal)>(?,?)"
+            params.extend(after.after)
+        rows = list(
+            conn.execute(
+                f"SELECT {_REFERENCE_COLUMNS} FROM execution_run_outputs WHERE {where} "
+                "ORDER BY output_node_id COLLATE BINARY,image_ordinal LIMIT ?",
+                (*params, limit + 1),
+            )
+        )
+        more = len(rows) > limit
+        rows = rows[:limit]
+        position = (
+            OutputPosition(
+                fingerprint, (rows[-1]["output_node_id"], rows[-1]["image_ordinal"])
+            )
+            if more
+            else None
+        )
+        return dict(
+            capture=capture,
+            retained_occurrence_count=count,
+            list_status=_list_status(capture),
+            capture_fingerprint=fingerprint,
+            items=[_receipt(run_id, row) for row in rows],
+            has_more=more,
+            next_position=position,
+        )
+
+    def get_history_occurrence(
+        self, run_id: str, node: str, ordinal: int
+    ) -> dict | None:
+        conn = self._history_connection()
+        capture, _ = _output_header(conn, run_id)
+        row = conn.execute(
+            f"SELECT {_REFERENCE_COLUMNS} FROM execution_run_outputs "
+            "WHERE run_id=? AND output_node_id=? COLLATE BINARY AND image_ordinal=?",
+            (run_id, node, ordinal),
+        ).fetchone()
+        return (
+            dict(capture=capture, receipt=_receipt(run_id, row))
+            if row is not None
+            else None
+        )

database/image_favorites.py

@@ -0,0 +1,152 @@
+"""Persisted image favorites, keyed by gallery root + relative path.
+
+The root key is the root's real, case-normalized path and the relative key is
+case-normalized the same way (see ``py/api/images.py``); both stay
+server-side. Folder names alone are not unique across roots, so a favorite
+follows the directory itself when the configured roots are reordered.
+Optional in old backups, required once a database is current (see
+``library_schema.py``).
+"""
+
+import re
+import sqlite3
+from typing import Iterable
+
+TABLE = "image_favorites"
+MAX_KEY_LENGTH = 4096
+_LOOKUP_CHUNK = 500
+_DDL = (
+    f"CREATE TABLE {TABLE} (root_key TEXT NOT NULL, relative_path TEXT NOT NULL, "
+    "created_at TEXT NOT NULL, PRIMARY KEY(root_key, relative_path))"
+)
+_AUTOINDEX = f"sqlite_autoindex_{TABLE}_1"
+_COLUMNS = (
+    ("root_key", "TEXT", 1, 1),
+    ("relative_path", "TEXT", 1, 2),
+    ("created_at", "TEXT", 1, 0),
+)
+
+
+def _compact(sql: str) -> str:
+    return re.sub(r"\s+", "", sql).lower()
+
+
+def image_favorites_schema_error(
+    conn: sqlite3.Connection, *, allow_missing: bool = True
+) -> str | None:
+    """Reject unsupported favorites schemas and invalid persisted values.
+
+    Reads only. A missing table is acceptable for backups that predate it
+    unless ``allow_missing`` is false.
+    """
+    objects = conn.execute(
+        "SELECT type,name,sql FROM sqlite_master WHERE name=? "
+        "OR (type IN ('index','trigger') AND tbl_name=?)",
+        (TABLE, TABLE),
+    ).fetchall()
+    if not objects:
+        return None if allow_missing else "Missing image favorites schema"
+    tables = [sql for kind, _name, sql in objects if kind == "table"]
+    others = {(kind, name, sql) for kind, name, sql in objects if kind != "table"}
+    if len(tables) != 1 or others != {("index", _AUTOINDEX, None)}:
+        return "Image favorites name is reserved by an unsupported schema object"
+    if _compact(tables[0] or "") != _compact(_DDL):
+        return "Image favorites has an invalid table definition"
+    columns = tuple(
+        (row[1], row[2].upper(), row[3], row[5])
+        for row in conn.execute(f"PRAGMA table_info({TABLE})")
+    )
+    if columns != _COLUMNS:
+        return "Image favorites has invalid columns"
+    # Bytes bound characters from above (UTF-8 is at most 4 bytes each).
+    limit = MAX_KEY_LENGTH * 4
+    if conn.execute(
+        f"SELECT 1 FROM {TABLE} WHERE typeof(root_key) != 'text' "
+        "OR typeof(relative_path) != 'text' OR typeof(created_at) != 'text' "
+        f"OR length(CAST(root_key AS BLOB)) > {limit} "
+        f"OR length(CAST(relative_path AS BLOB)) > {limit} "
+        "OR length(CAST(created_at AS BLOB)) > 64 LIMIT 1"
+    ).fetchone():
+        return "Image favorites contains invalid values"
+    for root, path in conn.execute(
+        f"SELECT CAST(root_key AS BLOB), CAST(relative_path AS BLOB) FROM {TABLE} "
+        f"WHERE length(CAST(root_key AS BLOB)) > {MAX_KEY_LENGTH} "
+        f"OR length(CAST(relative_path AS BLOB)) > {MAX_KEY_LENGTH}"
+    ):
+        for raw in (root, path):
+            if len(bytes(raw).decode("utf-8", "replace")) > MAX_KEY_LENGTH:
+                return "Image favorites contains invalid values"
+    return None
+
+
+def migrate_image_favorites(conn: sqlite3.Connection) -> None:
+    """Install the table atomically; validate before and after creating it."""
+    conn.execute("SAVEPOINT image_favorites_schema")
+    try:
+        error = image_favorites_schema_error(conn)
+        if error:
+            raise sqlite3.DatabaseError(error)
+        conn.execute(_DDL.replace("CREATE TABLE", "CREATE TABLE IF NOT EXISTS", 1))
+        error = image_favorites_schema_error(conn, allow_missing=False)
+        if error:
+            raise sqlite3.DatabaseError(error)
+        conn.execute("RELEASE image_favorites_schema")
+    except BaseException:
+        conn.execute("ROLLBACK TO image_favorites_schema")
+        conn.execute("RELEASE image_favorites_schema")
+        raise
+
+
+class ImageFavoriteOperations:
+    """PromptDatabase mixin: set, clear and look up image favorites."""
+
+    def set_image_favorite(self, root_key: str, relative_path: str) -> bool:
+        """Mark one file as a favorite; repeating it is a no-op. Returns True."""
+        from .models import utc_now_iso  # models imports this module's schema
+
+        with self.model.get_connection() as conn:
+            conn.execute(
+                "INSERT OR IGNORE INTO image_favorites VALUES (?, ?, ?)",
+                (root_key, relative_path, utc_now_iso()),
+            )
+        return True
+
+    def clear_image_favorite(self, root_key: str, relative_path: str) -> bool:
+        """Remove one favorite; True when a row was removed."""
+        with self.model.get_connection() as conn:
+            cursor = conn.execute(
+                "DELETE FROM image_favorites WHERE root_key=? AND relative_path=?",
+                (root_key, relative_path),
+            )
+        return cursor.rowcount > 0
+
+    def get_image_favorites(
+        self, root_key: str, relative_paths: Iterable[str]
+    ) -> set[str]:
+        """The subset of *relative_paths* favorited under *root_key*."""
+        wanted = list(dict.fromkeys(relative_paths))
+        found: set[str] = set()
+        conn = self.model.get_connection()
+        for start in range(0, len(wanted), _LOOKUP_CHUNK):
+            chunk = wanted[start : start + _LOOKUP_CHUNK]
+            marks = ",".join("?" * len(chunk))
+            rows = conn.execute(  # only "?" placeholders are interpolated
+                "SELECT relative_path FROM image_favorites WHERE root_key=? "
+                f"AND relative_path IN ({marks})",  # nosec B608
+                (root_key, *chunk),
+            )
+            found.update(path for (path,) in rows)
+        return found
+
+    def get_root_favorites(self, root_key: str) -> set[str]:
+        """Every relative key favorited under *root_key* (a primary-key range).
+
+        The Favorites listing (``favorites=1``) intersects these with the
+        root's scan, so its size follows the root's favorites, not its files.
+        """
+        conn = self.model.get_connection()
+        rows = conn.execute(
+            "SELECT relative_path FROM image_favorites WHERE root_key=?",
+            (root_key,),
+        )
+        return {path for (path,) in rows}

database/library.py

@@ -0,0 +1,630 @@
+"""Read-only positive-parent views over existing prompt and pair tables."""
+
+import datetime
+import sqlite3
+from collections.abc import Iterator
+from contextlib import contextmanager
+
+from .library_tokens import library_edit_token
+
+_ELIGIBLE = """(EXISTS(SELECT 1 FROM prompt_pairs ep WHERE ep.positive_prompt_id=p.id)
+ OR EXISTS(SELECT 1 FROM generated_images eg WHERE eg.prompt_id=p.id)
+ OR NOT EXISTS(SELECT 1 FROM prompt_pairs en WHERE en.negative_prompt_id=p.id)
+ OR EXISTS(SELECT 1 FROM prompt_library_state es WHERE es.prompt_id=p.id))"""
+_FROM = "prompts p LEFT JOIN prompt_library_state ls ON ls.prompt_id=p.id"
+# The characters JavaScript's String.prototype.trim() removes, so a title the
+# display treats as blank also falls back to the prompt text for name sorting.
+_TRIM_CODES = (9, 10, 11, 12, 13, 32, 160, 5760, *range(8192, 8203))
+_TRIM_CODES += (8232, 8233, 8239, 8287, 12288, 65279)
+_BLANK = f"char({','.join(map(str, _TRIM_CODES))})"
+_JS_TRIM = "".join(map(chr, _TRIM_CODES))
+# The Library's display title: a stored title, else the prompt's first line.
+DISPLAY_TITLE_LIMIT = 80
+
+
+def _utf16_units(text: str) -> int:
+    """JavaScript ``length``: characters beyond the BMP count as two units."""
+    return len(text) + sum(1 for ch in text if ord(ch) > 0xFFFF)
+
+
+def _utf16_prefix(text: str, units: int) -> str:
+    """The longest prefix within ``units`` UTF-16 units, never splitting a pair."""
+    used = 0
+    for index, ch in enumerate(text):
+        used += 2 if ord(ch) > 0xFFFF else 1
+        if used > units:
+            return text[:index]
+    return text
+
+
+def titled(stored_titles: tuple, text: object) -> tuple[str | None, str | None]:
+    """Return ``(title, source)``: a stored title, else the text's first line.
+
+    ``source`` is ``"stored"`` for the first non-blank stored title (blank as
+    in JavaScript's trim, like the Library view), ``"text"`` for the first line,
+    and None when both are empty. The first line is trimmed and, past
+    DISPLAY_TITLE_LIMIT UTF-16 units, cut with an ellipsis like the Library's
+    ``libraryTitle()`` (never inside a surrogate pair).
+    """
+    for title in stored_titles:
+        if isinstance(title, str) and title.strip(_JS_TRIM):
+            return title, "stored"
+    if not isinstance(text, str):
+        return None, None
+    first = text.split("\n")[0].strip(_JS_TRIM)
+    if _utf16_units(first) > DISPLAY_TITLE_LIMIT:
+        first = _utf16_prefix(first, DISPLAY_TITLE_LIMIT) + "\u2026"
+    return (first, "text") if first else (None, None)
+
+
+def display_title(stored_titles: tuple, text: object) -> str | None:
+    """The Library display title of ``titled``, without its source."""
+    return titled(stored_titles, text)[0]
+
+
+def utc_iso(value: object) -> str | None:
+    """A stored timestamp as UTC ISO text, or None when absent or unreadable.
+
+    Usage writes are already UTC ISO; a naive value (SQLite's CURRENT_TIMESTAMP
+    form) is read as UTC, and an offset is converted to UTC. One bad row must
+    not fail a whole Library read, so a value past the datetime range after
+    conversion (OverflowError) is unreadable too. Python 3.11+ fromisoformat
+    accepts more ISO forms than 3.10 (e.g. a 2-digit fraction); every form the
+    app writes (utc_now_iso, the 3.2.4 backfill, CURRENT_TIMESTAMP, Z) parses
+    the same on both, so only hand-edited values can differ by version.
+    """
+    if not isinstance(value, str):
+        return None
+    text = value.strip()
+    if text.endswith(("Z", "z")):
+        text = text[:-1] + "+00:00"
+    try:
+        stamp = datetime.datetime.fromisoformat(text)
+        if stamp.tzinfo is None:
+            stamp = stamp.replace(tzinfo=datetime.timezone.utc)
+        return stamp.astimezone(datetime.timezone.utc).isoformat()
+    except (ValueError, OverflowError):
+        return None
+
+
+_NAME = (
+    f"LOWER(CASE WHEN TRIM(COALESCE(ls.title,''),{_BLANK})!='' "
+    "THEN ls.title ELSE p.text END)"
+)
+_ORDERS = {
+    "newest": ("COALESCE(julianday(p.created_at),0)", "DESC"),
+    "oldest": ("COALESCE(julianday(p.created_at),0)", "ASC"),
+    "name": (_NAME, "ASC"),
+    "rating": ("COALESCE(p.rating,0)", "DESC"),
+}
+# Pair archive scopes for variant and image reads, as conditions on alias pp.
+ARCHIVE_SCOPES = {
+    "active": " AND pp.archived=0",
+    "archived": " AND pp.archived=1",
+    "all": "",
+}
+
+
+def _escape(value: str) -> str:
+    return value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
+
+
+def _limit(value: int) -> None:
+    if type(value) is not int or not 1 <= value <= 200:
+        raise ValueError("limit must be between 1 and 200")
+
+
+def _archive(value: str) -> str:
+    """Return the pair-archive condition for one scope (alias ``pp``)."""
+    if not isinstance(value, str) or value not in ARCHIVE_SCOPES:
+        raise ValueError("archive must be one of active, archived or all")
+    return ARCHIVE_SCOPES[value]
+
+
+def _filter(text: str, category: str | None, tags: list[str]) -> tuple[str, list]:
+    if not isinstance(text, str) or len(text) > 500:
+        raise ValueError("Search text must be a string of at most 500 characters")
+    if category is not None and (not isinstance(category, str) or len(category) > 200):
+        raise ValueError("Invalid library category")
+    if (
+        not isinstance(tags, list)
+        or len(tags) > 30
+        or any(not isinstance(tag, str) or not tag or len(tag) > 200 for tag in tags)
+    ):
+        raise ValueError("Invalid library tags")
+    parent, pair, parent_params, pair_params = [], ["pp.archived=0"], [], []
+    if text:
+        parent.append("(p.text LIKE ? ESCAPE '\\' OR ls.title LIKE ? ESCAPE '\\')")
+        parent_params.extend([f"%{_escape(text)}%"] * 2)
+        pair.append(
+            "(p.text LIKE ? ESCAPE '\\' OR ls.title LIKE ? ESCAPE '\\' "
+            "OR n.text LIKE ? ESCAPE '\\' OR pp.title LIKE ? ESCAPE '\\')"
+        )
+        pair_params.extend([f"%{_escape(text)}%"] * 4)
+    if category is not None:
+        parent.append("p.category=?")
+        pair.append("pp.category=?")
+        parent_params.append(category)
+        pair_params.append(category)
+    for tag in tags:
+        parent.append(
+            "EXISTS(SELECT 1 FROM prompt_tags pt JOIN tags t ON t.id=pt.tag_id "
+            "WHERE pt.prompt_id=p.id AND t.name=?)"
+        )
+        pair.append(
+            "EXISTS(SELECT 1 FROM prompt_pair_tags pt JOIN tags t ON t.id=pt.tag_id "
+            "WHERE pt.pair_id=pp.id AND t.name=?)"
+        )
+        parent_params.append(tag)
+        pair_params.append(tag)
+    condition = (
+        "("
+        + (" AND ".join(parent) or "1=1")
+        + (
+            ") OR EXISTS(SELECT 1 FROM prompt_pairs pp "
+            "LEFT JOIN prompts n ON n.id=pp.negative_prompt_id "
+            "WHERE pp.positive_prompt_id=p.id AND "
+        )
+        + " AND ".join(pair)
+        + ")"
+    )
+    return f"({_ELIGIBLE}) AND ({condition})", [*parent_params, *pair_params]
+
+
+class LibraryOperations:
+    """Group reusable negative variants without changing legacy prompt queries.
+
+    Each public read runs inside ``library_snapshot``; the ``_``-prefixed
+    helpers take that snapshot's connection and epoch so routes can share it.
+    """
+
+    @contextmanager
+    def library_snapshot(
+        self, *, include_ceiling: bool = False
+    ) -> Iterator[tuple[sqlite3.Connection, str, int | None]]:
+        """Hold one read transaction, reading the epoch (and ceiling) first."""
+        conn = self.model.get_connection()
+        if conn.in_transaction:
+            raise sqlite3.ProgrammingError("Library reads require their own snapshot")
+        conn.execute("BEGIN")
+        try:
+            row = conn.execute(
+                "SELECT storage_epoch,"
+                + (
+                    "(SELECT COALESCE(MAX(id),0) FROM prompts)"
+                    if include_ceiling
+                    else "NULL"
+                )
+                + " FROM execution_history_metadata WHERE singleton_id=1"
+            ).fetchone()
+            if row is None:
+                raise sqlite3.DatabaseError("Missing execution history metadata")
+            yield conn, row[0], row[1]
+        finally:
+            if conn.in_transaction:
+                conn.rollback()
+
+    def _library_parents(
+        self,
+        conn: sqlite3.Connection,
+        epoch: str,
+        where: str,
+        params: list,
+        suffix: str = "",
+        sort_expression: str | None = None,
+    ) -> list[dict]:
+        rows = conn.execute(
+            "SELECT p.id,p.text,ls.title,p.category,p.rating,p.notes,p.created_at,"
+            "p.run_count,p.last_used_at,"
+            + (f"{sort_expression} AS _sort," if sort_expression else "")
+            + "(SELECT COUNT(*) FROM prompt_pairs v WHERE v.positive_prompt_id=p.id "
+            "AND v.archived=0) AS variant_count,"
+            "(SELECT COUNT(*) FROM generated_images g "
+            "WHERE g.prompt_id=p.id) AS image_count "
+            "FROM " + _FROM + " WHERE " + where + suffix,
+            params,
+        ).fetchall()
+        result = [{**dict(row), "tags": []} for row in rows]
+        if result:
+            by_id = {row["id"]: row for row in result}
+            marks = ",".join("?" for _ in result)
+            for row in conn.execute(
+                "SELECT pt.prompt_id,t.name FROM prompt_tags pt "
+                "JOIN tags t ON t.id=pt.tag_id "
+                f"WHERE pt.prompt_id IN ({marks}) ORDER BY t.name",
+                list(by_id),
+            ):
+                by_id[row["prompt_id"]]["tags"].append(row["name"])
+        for row in result:
+            # Usage is shown, never compared: the token hashes owner fields only.
+            row["run_count"] = row["run_count"] if type(row["run_count"]) is int else 0
+            row["last_used_at"] = utc_iso(row["last_used_at"])
+            row["edit_token"] = library_edit_token(epoch, row)
+        return result
+
+    def _list_library(
+        self,
+        conn: sqlite3.Connection,
+        epoch: str,
+        ceiling: int,
+        *,
+        text: str = "",
+        category: str | None = None,
+        tags: list[str] | None = None,
+        sort: str = "newest",
+        limit: int = 40,
+        after: dict | None = None,
+    ) -> dict:
+        _limit(limit)
+        if sort not in _ORDERS:
+            raise ValueError("Unknown library sort order")
+        where, params = _filter(text, category, [] if tags is None else tags)
+        if after is not None:
+            if (
+                not isinstance(after, dict)
+                or set(after) != {"ceiling", "value", "id"}
+                or type(after["ceiling"]) is not int
+                or type(after["id"]) is not int
+            ):
+                raise ValueError("Invalid library cursor")
+            ceiling = after["ceiling"]
+        where += " AND p.id<=?"
+        params.append(ceiling)
+        # One SELECT counts the filtered parents and their active variants.
+        total, variant_total = conn.execute(
+            "SELECT COUNT(*),COALESCE(SUM((SELECT COUNT(*) FROM prompt_pairs v "
+            "WHERE v.positive_prompt_id=p.id AND v.archived=0)),0) FROM "
+            + _FROM
+            + " WHERE "
+            + where,
+            params,
+        ).fetchone()
+        expression, direction = _ORDERS[sort]
+        if after is not None:
+            operator = ">" if direction == "ASC" else "<"
+            where += (
+                f" AND ({expression}{operator}? OR "
+                f"({expression}=? AND p.id{operator}?))"
+            )
+            params.extend([after["value"], after["value"], after["id"]])
+        rows = self._library_parents(
+            conn,
+            epoch,
+            where,
+            [*params, limit + 1],
+            f" ORDER BY {expression} {direction},p.id {direction} LIMIT ?",
+            sort_expression=expression,
+        )
+        has_more = len(rows) > limit
+        rows = rows[:limit]
+        next_key = None
+        if has_more:
+            next_key = {
+                "ceiling": ceiling,
+                "value": rows[-1]["_sort"],
+                "id": rows[-1]["id"],
+            }
+        for row in rows:
+            row.pop("_sort")
+        return {
+            "library_epoch": epoch,
+            "items": rows,
+            "total": total,
+            "variant_total": variant_total,
+            "has_more": has_more,
+            "next_key": next_key,
+        }
+
+    def list_library(
+        self,
+        *,
+        text: str = "",
+        category: str | None = None,
+        tags: list[str] | None = None,
+        sort: str = "newest",
+        limit: int = 40,
+        after: dict | None = None,
+    ) -> dict:
+        with self.library_snapshot(include_ceiling=True) as (conn, epoch, ceiling):
+            return self._list_library(
+                conn,
+                epoch,
+                ceiling,
+                text=text,
+                category=category,
+                tags=tags,
+                sort=sort,
+                limit=limit,
+                after=after,
+            )
+
+    def _is_library_positive(self, conn: sqlite3.Connection, positive_id: int) -> bool:
+        return (
+            conn.execute(
+                "SELECT 1 FROM prompts p WHERE p.id=? AND " + _ELIGIBLE, (positive_id,)
+            ).fetchone()
+            is not None
+        )
+
+    def _get_library_positive(
+        self,
+        conn: sqlite3.Connection,
+        epoch: str,
+        positive_id: int,
+        *,
+        variant_limit: int = 40,
+        variant_after: int | None = None,
+        archive: str = "active",
+    ) -> dict | None:
+        _limit(variant_limit)
+        scope = _archive(archive)
+        if variant_after is not None and (
+            type(variant_after) is not int or variant_after <= 0
+        ):
+            raise ValueError("Invalid variant cursor")
+        rows = self._library_parents(
+            conn, epoch, _ELIGIBLE + " AND p.id=?", [positive_id]
+        )
+        if not rows:
+            return None
+        positive = rows[0]
+        positive["archived_variant_count"] = conn.execute(
+            "SELECT COUNT(*) FROM prompt_pairs "
+            "WHERE positive_prompt_id=? AND archived=1",
+            (positive_id,),
+        ).fetchone()[0]
+        variant_total = {
+            "active": positive["variant_count"],
+            "archived": positive["archived_variant_count"],
+            "all": positive["variant_count"] + positive["archived_variant_count"],
+        }[archive]
+        where = "pp.positive_prompt_id=?" + scope
+        params = [positive_id]
+        if variant_after is not None:
+            where += " AND pp.id<?"
+            params.append(variant_after)
+        variants = [
+            dict(row)
+            for row in conn.execute(
+                "SELECT pp.id,COALESCE(n.text,'') AS negative_text,"
+                "pp.category,pp.title,pp.archived,"
+                "pp.negative_prompt_id IS NULL AS explicit_empty,pp.run_count,"
+                "(SELECT COUNT(DISTINCT image_id) FROM image_prompt_pairs ip "
+                "WHERE ip.pair_id=pp.id) AS image_count "
+                "FROM prompt_pairs pp "
+                "LEFT JOIN prompts n ON n.id=pp.negative_prompt_id "
+                "WHERE " + where + " ORDER BY pp.id DESC LIMIT ?",
+                [*params, variant_limit + 1],
+            )
+        ]
+        variants_has_more = len(variants) > variant_limit
+        variants = variants[:variant_limit]
+        by_id = {row["id"]: row for row in variants}
+        for row in variants:
+            row["tags"] = []
+            row["archived"] = bool(row["archived"])
+            row["explicit_empty"] = bool(row["explicit_empty"])
+            # The history lane's "N runs" (#249): queue-hook pair uses, 0 when unset.
+            row["run_count"] = row["run_count"] if type(row["run_count"]) is int else 0
+        if by_id:
+            marks = ",".join("?" for _ in by_id)
+            for row in conn.execute(
+                "SELECT pt.pair_id,t.name FROM prompt_pair_tags pt "
+                "JOIN tags t ON t.id=pt.tag_id "
+                f"WHERE pt.pair_id IN ({marks}) ORDER BY t.name",
+                list(by_id),
+            ):
+                by_id[row["pair_id"]]["tags"].append(row["name"])
+        unassigned = conn.execute(
+            "SELECT COUNT(*) FROM generated_images g WHERE g.prompt_id=? "
+            "AND NOT EXISTS(SELECT 1 FROM image_prompt_pairs ip "
+            "WHERE ip.image_id=g.id)",
+            (positive_id,),
+        ).fetchone()[0]
+        return {
+            "library_epoch": epoch,
+            "positive": positive,
+            "variants": variants,
+            "unassigned_image_count": unassigned,
+            "variant_total": variant_total,
+            "variants_has_more": variants_has_more,
+            "variants_next_key": variants[-1]["id"] if variants_has_more else None,
+        }
+
+    def get_library_positive(
+        self,
+        positive_id: int,
+        *,
+        variant_limit: int = 40,
+        variant_after: int | None = None,
+        archive: str = "active",
+    ) -> dict | None:
+        with self.library_snapshot() as (conn, epoch, _ceiling):
+            return self._get_library_positive(
+                conn,
+                epoch,
+                positive_id,
+                variant_limit=variant_limit,
+                variant_after=variant_after,
+                archive=archive,
+            )
+
+    def _get_library_options(self, conn: sqlite3.Connection, epoch: str) -> dict:
+        rows = conn.execute(
+            "SELECT category FROM (SELECT p.category AS category "
+            "FROM prompts p WHERE "
+            + _ELIGIBLE
+            + (" UNION SELECT category FROM prompt_pairs " "WHERE archived=0) ")
+            + "WHERE category IS NOT NULL AND TRIM(category)!='' "
+            "ORDER BY category COLLATE NOCASE,category"
+        ).fetchall()
+        return {
+            "library_epoch": epoch,
+            "categories": [row["category"] for row in rows],
+        }
+
+    def get_library_options(self) -> dict:
+        with self.library_snapshot() as (conn, epoch, _ceiling):
+            return self._get_library_options(conn, epoch)
+
+    def get_library_previews(self, positive_ids: list[int]) -> dict[int, list[dict]]:
+        """Read at most two links per parent in one bounded page query."""
+        if (
+            not isinstance(positive_ids, list)
+            or len(positive_ids) > 200
+            or any(type(value) is not int or value <= 0 for value in positive_ids)
+        ):
+            raise ValueError("Invalid preview parent IDs")
+        if not positive_ids:
+            return {}
+        result = {value: [] for value in positive_ids}
+        marks = ",".join("?" for _ in positive_ids)
+        rows = self.model.get_connection().execute(
+            "WITH ranked AS (SELECT id,prompt_id,image_path,ROW_NUMBER() OVER "
+            "(PARTITION BY prompt_id ORDER BY id DESC) AS position "
+            f"FROM generated_images WHERE prompt_id IN ({marks})) "
+            "SELECT id,prompt_id,image_path FROM ranked WHERE position<=2 "
+            "ORDER BY prompt_id,id DESC",
+            positive_ids,
+        )
+        for row in rows:
+            result[row["prompt_id"]].append(dict(row))
+        return result
+
+    def _list_library_images(
+        self,
+        conn: sqlite3.Connection,
+        epoch: str,
+        positive_id: int,
+        *,
+        pair_id: int | None = None,
+        limit: int = 40,
+        after: int | None = None,
+        archive: str = "active",
+    ) -> dict:
+        _limit(limit)
+        scope = _archive(archive)
+        if after is not None and (type(after) is not int or after <= 0):
+            raise ValueError("Invalid image cursor")
+        if (
+            pair_id is not None
+            and not conn.execute(
+                "SELECT 1 FROM prompt_pairs pp WHERE pp.id=? "
+                "AND pp.positive_prompt_id=?" + scope,
+                (pair_id, positive_id),
+            ).fetchone()
+        ):
+            if conn.execute(
+                "SELECT 1 FROM prompt_pairs WHERE id=? AND positive_prompt_id=?",
+                (pair_id, positive_id),
+            ).fetchone():
+                raise ValueError("Variant is not in the requested archive scope")
+            raise ValueError("Variant does not belong to this positive prompt")
+        where, params = "g.prompt_id=?", [positive_id]
+        if pair_id is not None:
+            where += (
+                " AND EXISTS(SELECT 1 FROM image_prompt_pairs ip "
+                "WHERE ip.image_id=g.id AND ip.pair_id=?)"
+            )
+            params.append(pair_id)
+        total = conn.execute(
+            "SELECT COUNT(*) FROM generated_images g WHERE " + where, params
+        ).fetchone()[0]
+        if after is not None:
+            where += " AND g.id<?"
+            params.append(after)
+        rows = conn.execute(
+            "SELECT g.id,g.image_path FROM generated_images g WHERE "
+            + where
+            + " ORDER BY g.id DESC LIMIT ?",
+            [*params, limit + 1],
+        ).fetchall()
+        return {
+            "library_epoch": epoch,
+            "images": [dict(row) for row in rows[:limit]],
+            "total": total,
+            "has_more": len(rows) > limit,
+            "next_key": rows[limit - 1]["id"] if len(rows) > limit else None,
+        }
+
+    def list_library_images(
+        self,
+        positive_id: int,
+        *,
+        pair_id: int | None = None,
+        limit: int = 40,
+        after: int | None = None,
+        archive: str = "active",
+    ) -> dict:
+        with self.library_snapshot() as (conn, epoch, _ceiling):
+            return self._list_library_images(
+                conn,
+                epoch,
+                positive_id,
+                pair_id=pair_id,
+                limit=limit,
+                after=after,
+                archive=archive,
+            )
+
+    def export_library(self) -> dict:
+        """Read every portable owner and pair from one snapshot, writing nothing.
+
+        Local IDs order the rows internally and never appear in the document.
+        """
+        with self.library_snapshot() as (conn, _epoch, _ceiling):
+            owners = {}
+            order = []
+            for row in conn.execute(
+                "SELECT p.id,p.text,ls.title,p.category,p.rating,p.notes FROM "
+                + _FROM
+                + " WHERE "
+                + _ELIGIBLE
+                + " ORDER BY p.id"
+            ):
+                owners[row["id"]] = {
+                    "positive_text": row["text"],
+                    "title": row["title"],
+                    "category": row["category"],
+                    "tags": [],
+                    "rating": row["rating"],
+                    "notes": row["notes"],
+                    "variants": [],
+                }
+                order.append(row["id"])
+            for row in conn.execute(
+                "SELECT pt.prompt_id,t.name FROM prompt_tags pt "
+                "JOIN tags t ON t.id=pt.tag_id ORDER BY pt.prompt_id,t.name"
+            ):
+                if row["prompt_id"] in owners:
+                    owners[row["prompt_id"]]["tags"].append(row["name"])
+            variants = {}
+            for row in conn.execute(
+                "SELECT pp.id,pp.positive_prompt_id,"
+                "COALESCE(n.text,'') AS negative_text,"
+                "pp.title,pp.category,pp.rating,pp.notes,pp.archived "
+                "FROM prompt_pairs pp "
+                "LEFT JOIN prompts n ON n.id=pp.negative_prompt_id "
+                "ORDER BY pp.id"
+            ):
+                variant = {
+                    "negative_text": row["negative_text"],
+                    "title": row["title"],
+                    "category": row["category"],
+                    "tags": [],
+                    "rating": row["rating"],
+                    "notes": row["notes"],
+                    "archived": bool(row["archived"]),
+                }
+                variants[row["id"]] = variant
+                owners[row["positive_prompt_id"]]["variants"].append(variant)
+            for row in conn.execute(
+                "SELECT pt.pair_id,t.name FROM prompt_pair_tags pt "
+                "JOIN tags t ON t.id=pt.tag_id ORDER BY pt.pair_id,t.name"
+            ):
+                variants[row["pair_id"]]["tags"].append(row["name"])
+            return {
+                "schema_version": 1,
+                "kind": "prompt-manager-library",
+                "positives": [owners[key] for key in order],
+            }

database/library_commands.py

@@ -0,0 +1,310 @@
+"""Transactional native Library commands guarded by epoch and edit token."""
+
+import re
+import sqlite3
+
+from .library import _ELIGIBLE
+from .models import utc_now_iso
+
+try:
+    from ..utils.validators import (
+        validate_category,
+        validate_prompt_text,
+        validate_tags,
+    )
+except ImportError:
+    from utils.validators import validate_category, validate_prompt_text, validate_tags
+
+METADATA_FIELDS = frozenset({"title", "category", "tags", "rating", "notes"})
+MAX_TEXT_FIELD = 10000
+_UUID = re.compile(r"[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12}")
+_TOKEN = re.compile(r"[0-9a-f]{64}")
+# Parent metadata stored on the prompt row; title lives in prompt_library_state.
+_PROMPT_COLUMNS = ("category", "rating", "notes")
+
+
+class LibraryCommandError(Exception):
+    """A native command lost a race against storage or another writer.
+
+    ``code`` is a wire code (``metadata-conflict``, ``storage-changed`` or
+    ``vocabulary-conflict`` for a tag or category that changed since it was read),
+    ``epoch`` the storage epoch read inside the transaction and ``positive``
+    the current owner record when it may be disclosed.
+    """
+
+    def __init__(self, code: str, epoch: str, positive: dict | None = None):
+        super().__init__(code)
+        self.code = code
+        self.epoch = epoch
+        self.positive = positive
+
+
+def validate_library_metadata(metadata: dict, *, require_title: bool = False) -> None:
+    """Reject anything but a nonempty allowlisted parent metadata object."""
+    if not isinstance(metadata, dict) or not metadata:
+        raise ValueError("metadata must be a nonempty object")
+    unknown = set(metadata) - METADATA_FIELDS
+    if unknown:
+        raise ValueError("Unknown metadata fields: " + ", ".join(sorted(unknown)))
+    if require_title and (
+        not isinstance(metadata.get("title"), str) or not metadata["title"].strip()
+    ):
+        raise ValueError("title is required")
+    for name in ("title", "notes"):
+        value = metadata.get(name)
+        if value is not None and not isinstance(value, str):
+            raise ValueError(f"{name} must be a string or null")
+        if value is not None and len(value) > MAX_TEXT_FIELD:
+            raise ValueError(f"{name} is too long")
+    if metadata.get("category") is not None:
+        validate_category(metadata["category"])
+    if "tags" in metadata:
+        if not isinstance(metadata["tags"], list):
+            raise ValueError("tags must be a list of strings")
+        validate_tags(metadata["tags"])
+    rating = metadata.get("rating")
+    if rating is not None and (type(rating) is not int or not 1 <= rating <= 5):
+        raise ValueError("rating must be an integer between 1 and 5 or null")
+
+
+def normalize_library_metadata(metadata: dict) -> dict:
+    """Return validated metadata with category trimmed and blank stored as NULL."""
+    if "category" not in metadata:
+        return metadata
+    category = metadata["category"]
+    return {**metadata, "category": (category or "").strip() or None}
+
+
+def normalize_library_tags(tags: list[str]) -> list[str]:
+    """Trim and deduplicate validated tags, keeping first-seen order."""
+    return list(dict.fromkeys(tag.strip() for tag in tags))
+
+
+def _validate_epoch(expected_epoch: str) -> None:
+    if not isinstance(expected_epoch, str) or not _UUID.fullmatch(expected_epoch):
+        raise ValueError("expected_epoch must be a canonical UUID")
+
+
+def _validate_guards(positive_id: int, expected_epoch: str, if_match: str) -> None:
+    if type(positive_id) is not int or not 1 <= positive_id <= 2**63 - 1:
+        raise ValueError("Invalid positive prompt ID")
+    _validate_epoch(expected_epoch)
+    if not isinstance(if_match, str) or not _TOKEN.fullmatch(if_match):
+        raise ValueError("if_match must be 64 lowercase hexadecimal characters")
+
+
+def _validate_negative_text(negative_text: str) -> None:
+    """Require an explicit string; blank deliberately means the empty negative."""
+    if not isinstance(negative_text, str):
+        raise ValueError("negative_text must be explicit; use an empty string for none")
+    if negative_text.strip():
+        validate_prompt_text(negative_text)
+
+
+class LibraryCommandOperations:
+    """Connection-level parent commands mixed into PromptDatabase."""
+
+    def _library_epoch(self, conn: sqlite3.Connection, expected_epoch: str) -> str:
+        row = conn.execute(
+            "SELECT storage_epoch FROM execution_history_metadata WHERE singleton_id=1"
+        ).fetchone()
+        if row is None:
+            raise sqlite3.DatabaseError("Missing execution history metadata")
+        if row[0] != expected_epoch:
+            raise LibraryCommandError("storage-changed", row[0])
+        return row[0]
+
+    def _library_parent(
+        self, conn: sqlite3.Connection, epoch: str, positive_id: int
+    ) -> dict | None:
+        rows = self._library_parents(
+            conn, epoch, _ELIGIBLE + " AND p.id=?", [positive_id]
+        )
+        return rows[0] if rows else None
+
+    def _ensure_library_parent(self, conn: sqlite3.Connection, prompt_id: int) -> None:
+        """Register explicit parent intent once; an existing row is never touched."""
+        if conn.execute(
+            "SELECT 1 FROM prompt_library_state WHERE prompt_id=?", (prompt_id,)
+        ).fetchone():
+            return
+        conn.execute(
+            "INSERT INTO prompt_library_state(prompt_id,title) VALUES(?,NULL)",
+            (prompt_id,),
+        )
+
+    def _apply_library_metadata(
+        self, conn: sqlite3.Connection, current: dict, metadata: dict
+    ) -> None:
+        """Write only fields that differ; stamp ``prompts.updated_at`` once."""
+        prompt_id = current["id"]
+        changed = False
+        if "title" in metadata and metadata["title"] != current["title"]:
+            conn.execute(
+                "UPDATE prompt_library_state SET title=? WHERE prompt_id=?",
+                (metadata["title"], prompt_id),
+            )
+            changed = True
+        columns = {
+            name: metadata[name]
+            for name in _PROMPT_COLUMNS
+            if name in metadata and metadata[name] != current[name]
+        }
+        if "tags" in metadata:
+            tags = normalize_library_tags(metadata["tags"])
+            if set(tags) != set(current["tags"]):
+                self._sync_prompt_tags(conn, prompt_id, tags)
+                changed = True
+        if columns or changed:
+            columns["updated_at"] = utc_now_iso()
+            conn.execute(
+                "UPDATE prompts SET "
+                + ",".join(f"{name}=?" for name in columns)
+                + " WHERE id=?",
+                [*columns.values(), prompt_id],
+            )
+
+    def update_library_parent(
+        self,
+        positive_id: int,
+        *,
+        expected_epoch: str,
+        if_match: str,
+        metadata: dict,
+    ) -> dict | None:
+        """Apply explicit parent metadata if the epoch and edit token still match.
+
+        Returns ``{library_epoch, positive}`` read before commit, or None when
+        the positive is missing or not a Library parent. Raises
+        LibraryCommandError for a rotated epoch or a stale token.
+        """
+        _validate_guards(positive_id, expected_epoch, if_match)
+        validate_library_metadata(metadata)
+        metadata = normalize_library_metadata(metadata)
+        with self._pair_transaction() as conn:
+            epoch = self._library_epoch(conn, expected_epoch)
+            current = self._library_parent(conn, epoch, positive_id)
+            if current is None:
+                return None
+            if current["edit_token"] != if_match:
+                raise LibraryCommandError("metadata-conflict", epoch, current)
+            self._ensure_library_parent(conn, positive_id)
+            self._apply_library_metadata(conn, current, metadata)
+            positive = self._library_parent(conn, epoch, positive_id)
+            if positive is None:
+                raise sqlite3.DatabaseError("Library parent vanished during update")
+            return {"library_epoch": epoch, "positive": positive}
+
+    def _recipe_negative(
+        self, conn: sqlite3.Connection, negative_text: str
+    ) -> int | None:
+        if not negative_text.strip():
+            return None
+        return self._pair_component_outcome(conn, negative_text)[0]
+
+    def _recipe_receipt(
+        self,
+        conn: sqlite3.Connection,
+        epoch: str,
+        positive_id: int,
+        pair_id: int,
+        *,
+        positive_created: bool,
+        pair_created: bool,
+    ) -> dict:
+        """Read both owners inside the write transaction for the receipt."""
+        positive = self._library_parent(conn, epoch, positive_id)
+        pairs = self._read_pair_rows(conn, "pp.id = ?", [pair_id])
+        if positive is None or not pairs:
+            raise sqlite3.DatabaseError("Library recipe vanished during save")
+        return {
+            "library_epoch": epoch,
+            "positive": positive,
+            "pair": pairs[0],
+            "positive_created": positive_created,
+            "pair_created": pair_created,
+            # Parent metadata is applied only to a newly inserted positive.
+            "parent_metadata_applied": positive_created,
+        }
+
+    def create_library_recipe(
+        self,
+        positive_text: str,
+        negative_text: str,
+        *,
+        expected_epoch: str,
+        parent_metadata: dict,
+    ) -> dict:
+        """Create or reuse a positive and its ordered pair in one transaction.
+
+        Submitted parent metadata applies only when the positive component is
+        newly inserted; a reused component keeps its own metadata. New pairs get
+        empty pair metadata. Reused pairs keep theirs, including archive state.
+        Every success registers the positive as a Library parent.
+        """
+        _validate_epoch(expected_epoch)
+        validate_prompt_text(positive_text)
+        _validate_negative_text(negative_text)
+        validate_library_metadata(parent_metadata, require_title=True)
+        parent_metadata = normalize_library_metadata(parent_metadata)
+        with self._pair_transaction() as conn:
+            epoch = self._library_epoch(conn, expected_epoch)
+            positive_id, positive_created = self._pair_component_outcome(
+                conn, positive_text
+            )
+            negative_id = self._recipe_negative(conn, negative_text)
+            pair_id, pair_created = self._save_pair_components(
+                conn, positive_id, negative_id
+            )
+            self._ensure_library_parent(conn, positive_id)
+            if positive_created:
+                current = self._library_parent(conn, epoch, positive_id)
+                if current is None:
+                    raise sqlite3.DatabaseError("New Library parent is not eligible")
+                self._apply_library_metadata(conn, current, parent_metadata)
+            return self._recipe_receipt(
+                conn,
+                epoch,
+                positive_id,
+                pair_id,
+                positive_created=positive_created,
+                pair_created=pair_created,
+            )
+
+    def add_library_negative(
+        self,
+        positive_id: int,
+        negative_text: str,
+        *,
+        expected_epoch: str,
+        if_match: str,
+    ) -> dict | None:
+        """Create or reuse a pair under the exact stored positive.
+
+        The stored positive ID and text are used as they are; only the negative
+        is resolved. Returns None when the positive is missing or not a Library
+        parent, and raises LibraryCommandError for a rotated epoch or a stale
+        token, in both cases before any insertion.
+        """
+        _validate_guards(positive_id, expected_epoch, if_match)
+        _validate_negative_text(negative_text)
+        with self._pair_transaction() as conn:
+            epoch = self._library_epoch(conn, expected_epoch)
+            current = self._library_parent(conn, epoch, positive_id)
+            if current is None:
+                return None
+            if current["edit_token"] != if_match:
+                raise LibraryCommandError("metadata-conflict", epoch, current)
+            negative_id = self._recipe_negative(conn, negative_text)
+            pair_id, pair_created = self._save_pair_components(
+                conn, positive_id, negative_id
+            )
+            self._ensure_library_parent(conn, positive_id)
+            return self._recipe_receipt(
+                conn,
+                epoch,
+                positive_id,
+                pair_id,
+                positive_created=False,
+                pair_created=pair_created,
+            )

database/library_schema.py

@@ -0,0 +1,86 @@
+"""Optional-old, required-current schema for Library parent state."""
+
+import re
+import sqlite3
+
+TABLE = "prompt_library_state"
+MAX_TITLE_LENGTH = 10_000
+_DDL = (
+    f"CREATE TABLE {TABLE} (prompt_id INTEGER PRIMARY KEY, title TEXT, "
+    "FOREIGN KEY(prompt_id) REFERENCES prompts(id) ON DELETE CASCADE)"
+)
+_COLUMNS = (("prompt_id", "INTEGER", 0, 1), ("title", "TEXT", 0, 0))
+_FOREIGN_KEYS = {("prompt_id", "prompts", "id", "CASCADE")}
+
+
+def _compact(sql: str) -> str:
+    return re.sub(r"\s+", "", sql).lower()
+
+
+def library_schema_error(
+    conn: sqlite3.Connection, *, allow_missing: bool = True
+) -> str | None:
+    """Reject unsupported Library state schemas and invalid persisted values.
+
+    Reads only: nothing is ever repaired. A missing table is acceptable for
+    backups that predate it unless ``allow_missing`` is false.
+    """
+    objects = conn.execute(
+        "SELECT type,sql FROM sqlite_master WHERE name=? "
+        "OR (type IN ('index','trigger') AND tbl_name=?)",
+        (TABLE, TABLE),
+    ).fetchall()
+    if not objects:
+        return None if allow_missing else "Missing Library state schema"
+    tables = [sql for kind, sql in objects if kind == "table"]
+    if len(tables) != 1 or len(objects) != 1:
+        return "Library state name is reserved by an unsupported schema object"
+    if _compact(tables[0] or "") != _compact(_DDL):
+        return "Library state has an invalid table definition"
+    columns = tuple(
+        (row[1], row[2].upper(), row[3], row[5])
+        for row in conn.execute(f"PRAGMA table_info({TABLE})")
+    )
+    if columns != _COLUMNS:
+        return "Library state has invalid columns"
+    foreign_keys = {
+        (row[3], row[2], row[4], row[6])
+        for row in conn.execute(f"PRAGMA foreign_key_list({TABLE})")
+    }
+    if foreign_keys != _FOREIGN_KEYS:
+        return "Library state has invalid foreign key definitions"
+    if conn.execute(f"PRAGMA foreign_key_check({TABLE})").fetchone():
+        return "Library state contains foreign key violations"
+    # length() stops at an embedded NUL, so bound by bytes in SQL (characters
+    # never outnumber bytes, and UTF-8 uses at most 4 bytes per character) and
+    # count the few remaining candidates as decoded characters.
+    if conn.execute(
+        f"SELECT 1 FROM {TABLE} WHERE typeof(title) NOT IN ('null','text') "
+        f"OR length(CAST(title AS BLOB)) > {MAX_TITLE_LENGTH * 4} LIMIT 1"
+    ).fetchone():
+        return "Library state contains invalid titles"
+    for (raw,) in conn.execute(
+        f"SELECT CAST(title AS BLOB) FROM {TABLE} "
+        f"WHERE length(CAST(title AS BLOB)) > {MAX_TITLE_LENGTH}"
+    ):
+        if len(bytes(raw).decode("utf-8", "replace")) > MAX_TITLE_LENGTH:
+            return "Library state contains invalid titles"
+    return None
+
+
+def migrate_library_state(conn: sqlite3.Connection) -> None:
+    """Install the table atomically; validate before and after creating it."""
+    conn.execute("SAVEPOINT prompt_library_state_schema")
+    try:
+        error = library_schema_error(conn)
+        if error:
+            raise sqlite3.DatabaseError(error)
+        conn.execute(_DDL.replace("CREATE TABLE", "CREATE TABLE IF NOT EXISTS", 1))
+        error = library_schema_error(conn, allow_missing=False)
+        if error:
+            raise sqlite3.DatabaseError(error)
+        conn.execute("RELEASE prompt_library_state_schema")
+    except BaseException:
+        conn.execute("ROLLBACK TO prompt_library_state_schema")
+        conn.execute("RELEASE prompt_library_state_schema")
+        raise

database/library_tokens.py

@@ -0,0 +1,37 @@
+"""Narrow comparison tokens for a Library parent's own metadata projection."""
+
+import hashlib
+import json
+
+_DOMAIN = "prompt-manager-library-parent:v1\n"
+
+
+def canonical_token(domain: str, preimage: dict) -> str:
+    """SHA-256 of ``domain`` plus the canonical JSON of ``preimage``."""
+    payload = json.dumps(
+        preimage,
+        sort_keys=True,
+        separators=(",", ":"),
+        ensure_ascii=False,
+        allow_nan=False,
+    )
+    return hashlib.sha256((domain + payload).encode("utf-8")).hexdigest()
+
+
+def library_edit_token(epoch: str, positive: dict) -> str:
+    """Hash the parent's owner fields under the current storage epoch.
+
+    Timestamps, counts, pairs and previews are excluded, and an absent state
+    row projects the same null title as a default one.
+    """
+    preimage = {
+        "epoch": epoch,
+        "id": positive["id"],
+        "text": positive["text"],
+        "title": positive["title"],
+        "category": positive["category"],
+        "notes": positive["notes"],
+        "rating": positive["rating"],
+        "tags": sorted(set(positive["tags"])),
+    }
+    return canonical_token(_DOMAIN, preimage)

database/pair_ingestion.py

@@ -0,0 +1,136 @@
+"""Atomic metadata enrichment for a single known generation context."""
+
+import json
+import os
+from typing import Any, Dict
+
+from .models import normalize_image_path
+from .operations import _coerce_file_info
+
+try:
+    from ..utils.hashing import generate_prompt_hash
+except ImportError:
+    from utils.hashing import generate_prompt_hash
+
+
+def ingest_resolved_pair(
+    db: Any, image_path: str, metadata: Dict[str, Any], resolution: Dict[str, Any]
+) -> Dict[str, Any]:
+    state = resolution.get("state", "unresolved")
+    if state != "resolved":
+        return {"status": state, "pairs_added": 0, "links_added": 0}
+    positive, negative = resolution["positive"], resolution["negative"]
+    positive_text = positive.get("storage_text", positive["text"])
+    negative_text = negative.get("storage_text", negative["text"])
+    normalized = normalize_image_path(image_path)
+    provenance = {"resolver_version": 1, "positive": positive, "negative": negative}
+    encoded = json.dumps(
+        provenance, sort_keys=True, separators=(",", ":"), allow_nan=False
+    )
+    with db._pair_transaction() as conn:
+        images = conn.execute(
+            "SELECT gi.*,p.hash AS positive_hash FROM generated_images gi "
+            "JOIN prompts p ON p.id=gi.prompt_id "
+            "WHERE gi.file_path=? OR gi.image_path=?",
+            (normalized, image_path),
+        ).fetchall()
+        if len(images) > 1 or (
+            images and images[0]["positive_hash"] != generate_prompt_hash(positive_text)
+        ):
+            return {"status": "conflict", "pairs_added": 0, "links_added": 0}
+        existing = images[0] if images else None
+        if existing:
+            prior = conn.execute(
+                "SELECT n.hash AS negative_hash FROM image_prompt_pairs ip "
+                "JOIN prompt_pairs pp ON pp.id=ip.pair_id "
+                "LEFT JOIN prompts n ON n.id=pp.negative_prompt_id WHERE ip.image_id=?",
+                (existing["id"],),
+            ).fetchall()
+            expected = (
+                generate_prompt_hash(negative_text) if negative_text.strip() else None
+            )
+            if any(row["negative_hash"] != expected for row in prior):
+                return {"status": "conflict", "pairs_added": 0, "links_added": 0}
+        prior_pair = conn.execute(
+            "SELECT pp.id FROM prompt_pairs pp "
+            "JOIN prompts p ON p.id=pp.positive_prompt_id "
+            "LEFT JOIN prompts n ON n.id=pp.negative_prompt_id "
+            "WHERE p.hash=? AND n.hash IS ?",
+            (
+                generate_prompt_hash(positive_text),
+                generate_prompt_hash(negative_text) if negative_text.strip() else None,
+            ),
+        ).fetchone()
+        pair_id = db._save_pair(conn, positive_text, negative_text)
+        pair = conn.execute(
+            "SELECT * FROM prompt_pairs WHERE id=?", (pair_id,)
+        ).fetchone()
+        pairs_added = int(prior_pair is None)
+        file_size, width, height, image_format = _coerce_file_info(metadata)
+        snapshots = {
+            column: json.dumps(metadata[key], sort_keys=True, allow_nan=False)
+            for column, key in (
+                ("workflow_data", "workflow"),
+                ("prompt_metadata", "prompt"),
+                ("parameters", "parameters"),
+            )
+            if key in metadata
+        }
+        info = {
+            "file_size": file_size,
+            "width": width,
+            "height": height,
+            "format": image_format,
+        }
+        snapshots.update(
+            {key: value for key, value in info.items() if value is not None}
+        )
+        changed = False
+        if existing:
+            image_id = existing["id"]
+            updates = {
+                key: value for key, value in snapshots.items() if existing[key] != value
+            }
+            if updates:
+                conn.execute(
+                    "UPDATE generated_images SET "
+                    + ",".join(f"{key}=?" for key in updates)
+                    + " WHERE id=?",
+                    [*updates.values(), image_id],
+                )
+                changed = True
+        else:
+            columns = ["prompt_id", "image_path", "filename", "file_path", *snapshots]
+            values = [
+                pair["positive_prompt_id"],
+                image_path,
+                os.path.basename(image_path),
+                normalized,
+                *snapshots.values(),
+            ]
+            image_id = conn.execute(
+                "INSERT INTO generated_images("
+                + ",".join(columns)
+                + ") VALUES("
+                + ",".join("?" for _ in columns)
+                + ")",
+                values,
+            ).lastrowid
+        linked = conn.execute(
+            "SELECT 1 FROM image_prompt_pairs WHERE pair_id=? "
+            "AND image_id=? AND provenance=?",
+            (pair_id, image_id, encoded),
+        ).fetchone()
+        db._link_image_to_pair(conn, pair_id, image_id, provenance)
+        status = (
+            "added"
+            if existing is None
+            else "enriched" if changed or not linked else "unchanged"
+        )
+        return {
+            "status": status,
+            "pairs_added": pairs_added,
+            "links_added": int(not linked),
+            "pair_id": pair_id,
+            "image_id": image_id,
+        }

database/pairs.py

@@ -0,0 +1,433 @@
+"""Ordered reusable prompt pairs and their generation provenance."""
+
+import json
+import sqlite3
+from contextlib import contextmanager
+from typing import Any, Dict, Iterator, List, Optional
+
+from .models import pair_provenance_error, utc_now_iso
+
+try:
+    from ..utils.hashing import generate_prompt_hash
+except ImportError:
+    from utils.hashing import generate_prompt_hash
+
+
+class PairDependencyError(ValueError):
+    """An operation would change or remove content used by a saved pair."""
+
+
+def _validate_metadata(
+    title: Optional[str],
+    category: Optional[str],
+    tags: Optional[List[str]],
+    rating: Optional[int],
+    notes: Optional[str],
+) -> Optional[List[str]]:
+    for name, value in (("title", title), ("category", category), ("notes", notes)):
+        if value is not None and not isinstance(value, str):
+            raise ValueError(f"{name} must be a string")
+    if rating is not None and (type(rating) is not int or not 1 <= rating <= 5):
+        raise ValueError("Rating must be between 1 and 5")
+    if tags is None:
+        return None
+    if not isinstance(tags, list) or any(not isinstance(tag, str) for tag in tags):
+        raise ValueError("tags must be a list of strings")
+    return list(dict.fromkeys(tag.strip() for tag in tags if tag.strip()))
+
+
+class PairOperations:
+    """Connection-level pair operations mixed into PromptDatabase."""
+
+    @contextmanager
+    def _pair_transaction(self) -> Iterator[sqlite3.Connection]:
+        conn = self.model.get_connection()
+        if conn.in_transaction:
+            raise RuntimeError("Pair operations require their own transaction")
+        with conn:
+            conn.execute("BEGIN IMMEDIATE")
+            yield conn
+
+    def _ensure_prompts_unpaired(
+        self, conn: sqlite3.Connection, prompt_ids: List[int]
+    ) -> None:
+        if not prompt_ids:
+            return
+        placeholders = ",".join("?" for _ in prompt_ids)
+        row = conn.execute(
+            f"SELECT id FROM prompt_pairs WHERE positive_prompt_id IN ({placeholders}) "
+            f"OR negative_prompt_id IN ({placeholders}) LIMIT 1",
+            [*prompt_ids, *prompt_ids],
+        ).fetchone()
+        if row:
+            raise PairDependencyError(
+                f"Prompt belongs to saved pair {row['id']}; delete the pair first "
+                "or save changed text as a new prompt"
+            )
+
+    def _pair_component(self, conn: sqlite3.Connection, text: str) -> int:
+        return self._pair_component_outcome(conn, text)[0]
+
+    def _pair_component_outcome(
+        self, conn: sqlite3.Connection, text: str
+    ) -> tuple[int, bool]:
+        """Resolve a component by its dedup hash; report whether it was inserted.
+
+        The first stored spelling wins: the hash ignores outer whitespace and
+        case, internal whitespace stays significant. ``created`` comes from the
+        INSERT result under the caller's held write transaction.
+        """
+        prompt_hash = generate_prompt_hash(text)
+        now = utc_now_iso()
+        created = (
+            conn.execute(
+                "INSERT INTO prompts(text, hash, created_at, updated_at) "
+                "VALUES(?,?,?,?) ON CONFLICT(hash) DO NOTHING",
+                (text.strip(), prompt_hash, now, now),
+            ).rowcount
+            == 1
+        )
+        row = conn.execute(
+            "SELECT id FROM prompts WHERE hash = ?", (prompt_hash,)
+        ).fetchone()
+        return row["id"], created
+
+    def _sync_pair_tags(
+        self, conn: sqlite3.Connection, pair_id: int, tags: List[str]
+    ) -> None:
+        conn.execute("DELETE FROM prompt_pair_tags WHERE pair_id = ?", (pair_id,))
+        for tag_id in self._ensure_tags(conn, tags).values():
+            conn.execute(
+                "INSERT INTO prompt_pair_tags(pair_id,tag_id) VALUES(?,?)",
+                (pair_id, tag_id),
+            )
+
+    def save_prompt_pair(
+        self,
+        positive_text: str,
+        negative_text: str = "",
+        *,
+        title: Optional[str] = None,
+        category: Optional[str] = None,
+        tags: Optional[List[str]] = None,
+        rating: Optional[int] = None,
+        notes: Optional[str] = None,
+    ) -> int:
+        """Save an ordered pair atomically; repeat saves preserve its metadata."""
+        with self._pair_transaction() as conn:
+            return self._save_pair(
+                conn,
+                positive_text,
+                negative_text,
+                title=title,
+                category=category,
+                tags=tags,
+                rating=rating,
+                notes=notes,
+            )
+
+    def _save_pair(
+        self,
+        conn: sqlite3.Connection,
+        positive_text: str,
+        negative_text: str = "",
+        *,
+        title: Optional[str] = None,
+        category: Optional[str] = None,
+        tags: Optional[List[str]] = None,
+        rating: Optional[int] = None,
+        notes: Optional[str] = None,
+    ) -> int:
+        """Save an ordered pair atomically; repeat saves preserve its metadata."""
+        if not isinstance(positive_text, str) or not positive_text.strip():
+            raise ValueError("Positive prompt text cannot be empty")
+        if not isinstance(negative_text, str):
+            raise ValueError("Negative prompt text must be a string; use '' for empty")
+        # Validate metadata before any component insertion.
+        _validate_metadata(title, category, tags, rating, notes)
+        positive = self._pair_component(conn, positive_text)
+        negative = (
+            self._pair_component(conn, negative_text) if negative_text.strip() else None
+        )
+        return self._save_pair_components(
+            conn,
+            positive,
+            negative,
+            title=title,
+            category=category,
+            tags=tags,
+            rating=rating,
+            notes=notes,
+        )[0]
+
+    def _save_pair_components(
+        self,
+        conn: sqlite3.Connection,
+        positive_id: int,
+        negative_id: Optional[int],
+        *,
+        title: Optional[str] = None,
+        category: Optional[str] = None,
+        tags: Optional[List[str]] = None,
+        rating: Optional[int] = None,
+        notes: Optional[str] = None,
+    ) -> tuple[int, bool]:
+        """Create or reuse the ordered pair of existing component IDs.
+
+        Returns ``(pair_id, created)``. A reused pair keeps all of its metadata
+        and archive state; metadata applies only to a newly inserted pair.
+        """
+        tags = _validate_metadata(title, category, tags, rating, notes)
+        existing = conn.execute(
+            "SELECT id FROM prompt_pairs WHERE positive_prompt_id = ? "
+            "AND negative_prompt_id IS ?",
+            (positive_id, negative_id),
+        ).fetchone()
+        if existing:
+            return existing["id"], False
+        now = utc_now_iso()
+        pair_id = conn.execute(
+            "INSERT INTO prompt_pairs(positive_prompt_id,negative_prompt_id,title,"
+            "category,rating,notes,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?)",
+            (positive_id, negative_id, title, category, rating, notes, now, now),
+        ).lastrowid
+        if tags:
+            self._sync_pair_tags(conn, pair_id, tags)
+        return pair_id, True
+
+    def _read_pair_rows(
+        self, conn: sqlite3.Connection, where: str, params: List[Any], suffix: str = ""
+    ) -> List[Dict[str, Any]]:
+        rows = conn.execute(
+            "SELECT pp.*, p.text AS positive_text, "
+            "COALESCE(n.text,'') AS negative_text,"
+            "(SELECT COUNT(DISTINCT image_id) FROM image_prompt_pairs ip "
+            "WHERE ip.pair_id=pp.id) AS image_count FROM prompt_pairs pp "
+            "JOIN prompts p ON p.id=pp.positive_prompt_id "
+            "LEFT JOIN prompts n ON n.id=pp.negative_prompt_id WHERE " + where + suffix,
+            params,
+        ).fetchall()
+        pairs = [dict(row) for row in rows]
+        if not pairs:
+            return pairs
+        by_id = {pair["id"]: pair for pair in pairs}
+        for pair in pairs:
+            pair["archived"] = bool(pair["archived"])
+            pair["tags"] = []
+        placeholders = ",".join("?" for _ in by_id)
+        for row in conn.execute(
+            "SELECT pt.pair_id,t.name FROM prompt_pair_tags pt JOIN tags t ON "
+            f"t.id=pt.tag_id WHERE pt.pair_id IN ({placeholders}) ORDER BY t.name",
+            list(by_id),
+        ):
+            by_id[row["pair_id"]]["tags"].append(row["name"])
+        return pairs
+
+    def get_prompt_pair(self, pair_id: int) -> Optional[Dict[str, Any]]:
+        """Return components, pair metadata and image count, including archives."""
+        rows = self._read_pair_rows(self.model.get_connection(), "pp.id = ?", [pair_id])
+        return rows[0] if rows else None
+
+    def list_prompt_pairs(
+        self,
+        *,
+        text: Optional[str] = None,
+        category: Optional[str] = None,
+        tags: Optional[List[str]] = None,
+        include_archived: bool = False,
+        limit: int = 50,
+        offset: int = 0,
+        sort: str = "newest",
+    ) -> Dict[str, Any]:
+        """List pairs with all-tag filtering and stable paginated ordering."""
+        orders = {
+            "newest": "pp.id DESC",
+            "last_used_desc": "julianday(pp.last_used_at) DESC, pp.id DESC",
+        }
+        if sort not in orders:
+            raise ValueError("Unknown pair sort order")
+        if type(limit) is not int or not 1 <= limit <= 500:
+            raise ValueError("limit must be between 1 and 500")
+        if type(offset) is not int or offset < 0:
+            raise ValueError("offset must be a nonnegative integer")
+        tags = _validate_metadata(None, category, tags, None, None)
+        if text is not None and not isinstance(text, str):
+            raise ValueError("text must be a string")
+        clauses, params = ["1=1"], []
+        if not include_archived:
+            clauses.append("pp.archived=0")
+        if category is not None:
+            clauses.append("pp.category = ?")
+            params.append(category)
+        if text:
+            clauses.append("(p.text LIKE ? OR n.text LIKE ? OR pp.title LIKE ?)")
+            params.extend([f"%{text}%"] * 3)
+        for tag in tags or []:
+            clauses.append(
+                "EXISTS(SELECT 1 FROM prompt_pair_tags pt JOIN tags t "
+                "ON t.id=pt.tag_id WHERE pt.pair_id=pp.id AND t.name=?)"
+            )
+            params.append(tag)
+        where = " AND ".join(clauses)
+        conn = self.model.get_connection()
+        total = conn.execute(
+            "SELECT COUNT(*) FROM prompt_pairs pp "
+            "JOIN prompts p ON p.id=pp.positive_prompt_id "
+            "LEFT JOIN prompts n ON n.id=pp.negative_prompt_id WHERE " + where,
+            params,
+        ).fetchone()[0]
+        pairs = self._read_pair_rows(
+            conn,
+            where,
+            [*params, limit, offset],
+            f" ORDER BY {orders[sort]} LIMIT ? OFFSET ?",
+        )
+        return {
+            "pairs": pairs,
+            "total": total,
+            "limit": limit,
+            "offset": offset,
+            "has_more": offset + len(pairs) < total,
+        }
+
+    def update_prompt_pair_metadata(
+        self,
+        pair_id: int,
+        *,
+        title: Optional[str] = None,
+        category: Optional[str] = None,
+        tags: Optional[List[str]] = None,
+        rating: Optional[int] = None,
+        notes: Optional[str] = None,
+    ) -> bool:
+        """Update pair metadata explicitly; None preserves and empty values clear."""
+        tags = _validate_metadata(title, category, tags, rating, notes)
+        changes = {
+            key: value
+            for key, value in (
+                ("title", title),
+                ("category", category),
+                ("rating", rating),
+                ("notes", notes),
+            )
+            if value is not None
+        }
+        if not changes and tags is None:
+            return False
+        with self._pair_transaction() as conn:
+            if not conn.execute(
+                "SELECT 1 FROM prompt_pairs WHERE id=?", (pair_id,)
+            ).fetchone():
+                return False
+            changes["updated_at"] = utc_now_iso()
+            conn.execute(
+                "UPDATE prompt_pairs SET "
+                + ",".join(f"{key}=?" for key in changes)
+                + " WHERE id=?",
+                [*changes.values(), pair_id],
+            )
+            if tags is not None:
+                self._sync_pair_tags(conn, pair_id, tags)
+            return True
+
+    def archive_prompt_pair(self, pair_id: int, archived: bool = True) -> bool:
+        """Hide a reusable pair without changing components or generation history."""
+        if type(archived) is not bool:
+            raise ValueError("archived must be a boolean")
+        with self._pair_transaction() as conn:
+            return (
+                conn.execute(
+                    "UPDATE prompt_pairs SET archived=?,updated_at=? WHERE id=?",
+                    (int(archived), utc_now_iso(), pair_id),
+                ).rowcount
+                > 0
+            )
+
+    def delete_prompt_pair(self, pair_id: int) -> bool:
+        """Delete unused pair metadata; generated pairs must be archived instead."""
+        with self._pair_transaction() as conn:
+            if conn.execute(
+                "SELECT 1 FROM image_prompt_pairs WHERE pair_id=? LIMIT 1", (pair_id,)
+            ).fetchone():
+                raise PairDependencyError(
+                    "Pair has generated images; archive it instead"
+                )
+            return (
+                conn.execute("DELETE FROM prompt_pairs WHERE id=?", (pair_id,)).rowcount
+                > 0
+            )
+
+    def record_prompt_pair_use(self, pair_id: int, times: int = 1) -> bool:
+        """Count pair uses independently of standalone prompt usage."""
+        if type(times) is not int or times < 1:
+            raise ValueError("times must be a positive integer")
+        with self._pair_transaction() as conn:
+            return (
+                conn.execute(
+                    "UPDATE prompt_pairs SET run_count=run_count+?,last_used_at=? "
+                    "WHERE id=?",
+                    (times, utc_now_iso(), pair_id),
+                ).rowcount
+                > 0
+            )
+
+    def link_image_to_pair(
+        self, pair_id: int, image_id: int, provenance: Dict[str, Any]
+    ) -> int:
+        """Associate an image with immutable run provenance, idempotently."""
+        with self._pair_transaction() as conn:
+            return self._link_image_to_pair(conn, pair_id, image_id, provenance)
+
+    def _link_image_to_pair(
+        self,
+        conn: sqlite3.Connection,
+        pair_id: int,
+        image_id: int,
+        provenance: Dict[str, Any],
+    ) -> int:
+        """Associate an image with immutable run provenance, idempotently."""
+        if not isinstance(provenance, dict):
+            raise ValueError("provenance must be an object")
+        try:
+            encoded = json.dumps(
+                provenance, sort_keys=True, separators=(",", ":"), allow_nan=False
+            )
+        except (TypeError, ValueError, RecursionError) as error:
+            raise ValueError("provenance must contain JSON values") from error
+        error = pair_provenance_error(encoded)
+        if error:
+            raise ValueError(error)
+        pair = conn.execute(
+            "SELECT positive_prompt_id FROM prompt_pairs WHERE id=?", (pair_id,)
+        ).fetchone()
+        image = conn.execute(
+            "SELECT prompt_id FROM generated_images WHERE id=?", (image_id,)
+        ).fetchone()
+        if not pair or not image:
+            raise ValueError("Pair and image must exist")
+        if pair["positive_prompt_id"] != image["prompt_id"]:
+            raise ValueError("Image must belong to the pair's positive prompt")
+        conn.execute(
+            "INSERT INTO image_prompt_pairs(pair_id,image_id,provenance,created_at) "
+            "VALUES(?,?,?,?) ON CONFLICT(pair_id,image_id,provenance) DO NOTHING",
+            (pair_id, image_id, encoded, utc_now_iso()),
+        )
+        return conn.execute(
+            "SELECT id FROM image_prompt_pairs "
+            "WHERE pair_id=? AND image_id=? AND provenance=?",
+            (pair_id, image_id, encoded),
+        ).fetchone()["id"]
+
+    def get_image_prompt_pairs(self, image_id: int) -> List[Dict[str, Any]]:
+        """Return every recorded pair association with decoded provenance."""
+        rows = (
+            self.model.get_connection()
+            .execute(
+                "SELECT * FROM image_prompt_pairs WHERE image_id=? ORDER BY id",
+                (image_id,),
+            )
+            .fetchall()
+        )
+        return [
+            {**dict(row), "provenance": json.loads(row["provenance"])} for row in rows
+        ]

database/saved_views.py

@@ -0,0 +1,334 @@
+"""Persisted saved views (D16): a named Images or Library toolbar state.
+
+A view's state holds exactly its page's toolbar values and its Refine-this-view
+refinements, whitelisted per page and stored as compact JSON of at most 4 KB.
+
+Writes (the API's POST and PATCH) validate strictly. Startup and restore check
+only the structure (the table, its columns and each row's types, sizes and JSON
+object), never the value rules: a row saved under older rules must not stop the
+database from opening. Startup is more tolerant still: a row whose state is not
+a JSON object is logged by id, left in place and skipped by every read, while
+restore rejects such a backup. Reads normalize instead, so a stale row opens with its
+usable values and defaults for the rest. Optional in old backups, required once
+a database is current (see ``library_schema.py``).
+"""
+
+import json
+import logging
+import re
+import sqlite3
+from typing import Optional
+
+# The models logger's parent ("prompt_manager.database.models" logs there too).
+_LOGGER = logging.getLogger("prompt_manager.database.saved_views")
+
+TABLE = "saved_views"
+PAGES = ("images", "library")
+# The schema check's literal SQL repeats the name and state caps (names in
+# bytes, at most 4 per character); keep them in step.
+MAX_NAME_LENGTH = 100
+MAX_STATE_BYTES = 4096
+MAX_TAGS = 20
+MAX_TAG_LENGTH = 50
+_IMAGE_SORTS = ("newest", "oldest", "name", "name-desc", "rating")
+_LIBRARY_SORTS = ("newest", "oldest", "name", "rating")
+_MEDIA = ("all", "image", "video")
+_CONTROL = re.compile(r"[\x00-\x1f\x7f]")
+_SOURCE = re.compile(r"all|0|[1-9][0-9]{0,8}")
+FALLBACK_NAME = "Saved view"
+
+
+class SavedViewInvalid(ValueError):
+    """A request's page, name or state breaks the write rules (a 400)."""
+
+
+# (kind, limit-or-choices, default) per state key; "text" values are
+# length-capped. The defaults match the client's (saved-views.mjs FIELDS).
+_STATE_FIELDS = {
+    "images": {
+        "q": ("text", 500, ""),
+        "category": ("text", 100, ""),
+        "sort": ("choice", _IMAGE_SORTS, "newest"),
+        "source": ("source", None, "all"),
+        "folder": ("text", 1024, ""),
+        "media": ("choice", _MEDIA, "all"),
+        "tags": ("tags", None, ()),
+        # The Favorites page's filter (#249 D51): a view saved there reopens it.
+        "favorites": ("bool", None, False),
+    },
+    "library": {
+        "q": ("text", 500, ""),
+        "category": ("text", 100, ""),
+        "sort": ("choice", _LIBRARY_SORTS, "newest"),
+        "tags": ("tags", None, ()),
+    },
+}
+
+_DDL = (
+    f"CREATE TABLE {TABLE} (id INTEGER PRIMARY KEY AUTOINCREMENT, "
+    "page TEXT NOT NULL CHECK(page IN ('images','library')), "
+    "name TEXT NOT NULL, state TEXT NOT NULL, position INTEGER NOT NULL, "
+    "created_at TEXT NOT NULL, updated_at TEXT NOT NULL)"
+)
+_COLUMNS = (
+    ("id", "INTEGER", 0, 1),
+    ("page", "TEXT", 1, 0),
+    ("name", "TEXT", 1, 0),
+    ("state", "TEXT", 1, 0),
+    ("position", "INTEGER", 1, 0),
+    ("created_at", "TEXT", 1, 0),
+    ("updated_at", "TEXT", 1, 0),
+)
+_SELECT = "SELECT id, page, name, state FROM saved_views"
+
+
+def _compact(sql: str) -> str:
+    return re.sub(r"\s+", "", sql).lower()
+
+
+def validate_name(name: object) -> str:
+    """The trimmed name, 1 to 100 characters without control characters."""
+    if not isinstance(name, str):
+        raise SavedViewInvalid("Name must be a string")
+    trimmed = name.strip()
+    if not 1 <= len(trimmed) <= MAX_NAME_LENGTH:
+        raise SavedViewInvalid(f"Name must be 1 to {MAX_NAME_LENGTH} characters")
+    if _CONTROL.search(trimmed):
+        raise SavedViewInvalid("Name contains control characters")
+    return trimmed
+
+
+def _validate_value(key: str, kind: str, limit, value: object) -> None:
+    """Raise ``SavedViewInvalid`` when *value* breaks *key*'s write rule."""
+    if kind == "tags":
+        if not isinstance(value, list) or len(value) > MAX_TAGS:
+            raise SavedViewInvalid(
+                f"State '{key}' must be a list of at most {MAX_TAGS}"
+            )
+        for tag in value:
+            if (
+                not isinstance(tag, str)
+                or not tag.strip()
+                or len(tag) > MAX_TAG_LENGTH
+                or "," in tag
+                or _CONTROL.search(tag)
+            ):
+                raise SavedViewInvalid(f"State '{key}' holds an invalid tag")
+        return
+    if kind == "bool":
+        if not isinstance(value, bool):
+            raise SavedViewInvalid(f"State '{key}' must be true or false")
+        return
+    if not isinstance(value, str):
+        raise SavedViewInvalid(f"State '{key}' must be a string")
+    if kind == "choice" and value not in limit:
+        raise SavedViewInvalid(f"State '{key}' is not a supported value")
+    if kind == "source" and not _SOURCE.fullmatch(value):
+        raise SavedViewInvalid(f"State '{key}' must be 'all' or a source index")
+    if kind == "text" and (len(value) > limit or _CONTROL.search(value)):
+        raise SavedViewInvalid(f"State '{key}' is too long or has control characters")
+
+
+def encode_state(page: object, state: object) -> str:
+    """Validate *state* for *page* and return its compact JSON (at most 4 KB)."""
+    if page not in PAGES:
+        raise SavedViewInvalid("Page must be 'images' or 'library'")
+    fields = _STATE_FIELDS[page]
+    if not isinstance(state, dict):
+        raise SavedViewInvalid("State must be an object")
+    if set(state) != set(fields):
+        raise SavedViewInvalid(
+            f"State for {page} must contain exactly " + ", ".join(sorted(fields))
+        )
+    for key, (kind, limit, _default) in fields.items():
+        _validate_value(key, kind, limit, state[key])
+    encoded = json.dumps(
+        {key: state[key] for key in fields}, ensure_ascii=False, separators=(",", ":")
+    )
+    if len(encoded.encode("utf-8")) > MAX_STATE_BYTES:
+        raise SavedViewInvalid("State must encode to at most 4 KB")
+    return encoded
+
+
+def _decode_object(raw: object) -> Optional[dict]:
+    """The JSON object stored in *raw*, or None when it holds no object."""
+    if isinstance(raw, dict):
+        return raw
+    if not isinstance(raw, str):
+        return None
+    try:
+        decoded = json.loads(raw)
+    except (ValueError, RecursionError):
+        return None
+    return decoded if isinstance(decoded, dict) else None
+
+
+def normalize_state(page: str, raw: object) -> dict:
+    """The page's full state from a stored row, never raising.
+
+    Every key of the page's current whitelist is present: a stored value that
+    passes today's write rule is kept, anything else (missing, invalid, or a
+    row that is not a JSON object) takes the key's default, and keys the
+    whitelist no longer has are dropped. An unknown page has no keys.
+    """
+    fields = _STATE_FIELDS.get(page, {})
+    stored = _decode_object(raw) or {}
+    state = {}
+    for key, (kind, limit, default) in fields.items():
+        value = stored.get(key, default)
+        try:
+            _validate_value(key, kind, limit, value)
+        except SavedViewInvalid:
+            value = default
+        state[key] = list(value) if kind == "tags" else value
+    return state
+
+
+def normalize_name(name: object) -> str:
+    """A stored name made displayable: trimmed, capped, never empty."""
+    text = _CONTROL.sub("", name) if isinstance(name, str) else ""
+    trimmed = text.strip()[:MAX_NAME_LENGTH].strip()
+    return trimmed or FALLBACK_NAME
+
+
+def saved_views_schema_error(
+    conn: sqlite3.Connection,
+    *,
+    allow_missing: bool = True,
+    check_states: bool = True,
+) -> Optional[str]:
+    """Reject unsupported saved-views schemas and structurally invalid rows.
+
+    Reads only. Checks the table, its columns, each row's types and sizes and
+    that each state is a JSON object; never the value rules, which change
+    between versions and are normalized on read instead. A missing table is
+    acceptable for backups that predate it unless ``allow_missing`` is false.
+    ``check_states=False`` (live startup) skips the JSON-object check; such
+    rows are reported by ``unusable_view_ids`` and skipped on read.
+    """
+    objects = conn.execute(
+        "SELECT type,name,sql FROM sqlite_master WHERE name=? "
+        "OR (type IN ('index','trigger') AND tbl_name=?)",
+        (TABLE, TABLE),
+    ).fetchall()
+    if not objects:
+        return None if allow_missing else "Missing saved views schema"
+    tables = [sql for kind, _name, sql in objects if kind == "table"]
+    if len(tables) != 1 or len(objects) != 1:
+        return "Saved views name is reserved by an unsupported schema object"
+    if _compact(tables[0] or "") != _compact(_DDL):
+        return "Saved views has an invalid table definition"
+    columns = tuple(
+        (row[1], row[2].upper(), row[3], row[5])
+        for row in conn.execute(f"PRAGMA table_info({TABLE})")
+    )
+    if columns != _COLUMNS:
+        return "Saved views has invalid columns"
+    # Literal SQL: the table name and limits are constants of this module.
+    if conn.execute(
+        "SELECT 1 FROM saved_views WHERE typeof(page) != 'text' "
+        "OR page NOT IN ('images','library') "
+        "OR typeof(name) != 'text' OR typeof(state) != 'text' "
+        "OR typeof(position) != 'integer' OR typeof(created_at) != 'text' "
+        "OR typeof(updated_at) != 'text' "
+        "OR length(CAST(name AS BLOB)) > 400 "
+        "OR length(CAST(state AS BLOB)) > 4096 "
+        "OR length(CAST(created_at AS BLOB)) > 64 "
+        "OR length(CAST(updated_at AS BLOB)) > 64 LIMIT 1"
+    ).fetchone():
+        return "Saved views contains invalid values"
+    if check_states and unusable_view_ids(conn):
+        return "Saved views contains invalid values"
+    return None
+
+
+def unusable_view_ids(conn: sqlite3.Connection) -> list[int]:
+    """Ids of rows whose state is not a JSON object (never listed)."""
+    return [
+        view_id
+        for view_id, state in conn.execute("SELECT id, state FROM saved_views")
+        if _decode_object(state) is None
+    ]
+
+
+def migrate_saved_views(conn: sqlite3.Connection) -> None:
+    """Install the table atomically; validate before and after creating it."""
+    conn.execute("SAVEPOINT saved_views_schema")
+    try:
+        error = saved_views_schema_error(conn, check_states=False)
+        if error:
+            raise sqlite3.DatabaseError(error)
+        conn.execute(_DDL.replace("CREATE TABLE", "CREATE TABLE IF NOT EXISTS", 1))
+        error = saved_views_schema_error(conn, allow_missing=False, check_states=False)
+        if error:
+            raise sqlite3.DatabaseError(error)
+        # A corrupt state never stops the live database from opening: the row
+        # stays (no data is destroyed) and reads skip it. Only its id is logged.
+        for view_id in unusable_view_ids(conn):
+            _LOGGER.warning(
+                "Saved view id %d has an unreadable state and is hidden", view_id
+            )
+        conn.execute("RELEASE saved_views_schema")
+    except BaseException:
+        conn.execute("ROLLBACK TO saved_views_schema")
+        conn.execute("RELEASE saved_views_schema")
+        raise
+
+
+def _view(row) -> dict:
+    view_id, page, name, state = row
+    return {
+        "id": view_id,
+        "page": page,
+        "name": normalize_name(name),
+        "state": normalize_state(page, state),
+    }
+
+
+class SavedViewOperations:
+    """PromptDatabase mixin: list, create, rename and delete saved views."""
+
+    def list_saved_views(self) -> list[dict]:
+        """Every view, in list order (position, then creation)."""
+        rows = self.model.get_connection().execute(_SELECT + " ORDER BY position, id")
+        # A row whose state is no JSON object (logged at startup) is never listed.
+        return [_view(row) for row in rows if _decode_object(row[3]) is not None]
+
+    def create_saved_view(self, page: object, name: object, state: object) -> dict:
+        """Validate and append one view at the end of the list."""
+        from .models import utc_now_iso  # models imports this module's schema
+
+        encoded = encode_state(page, state)
+        trimmed = validate_name(name)
+        now = utc_now_iso()
+        with self.model.get_connection() as conn:
+            cursor = conn.execute(
+                "INSERT INTO saved_views (page, name, state, position, "
+                "created_at, updated_at) SELECT ?, ?, ?, "
+                "COALESCE(MAX(position), -1) + 1, ?, ? FROM saved_views",
+                (page, trimmed, encoded, now, now),
+            )
+            row = conn.execute(
+                _SELECT + " WHERE id = ?", (cursor.lastrowid,)
+            ).fetchone()
+        return _view(row)
+
+    def rename_saved_view(self, view_id: int, name: object) -> Optional[dict]:
+        """The renamed view, or None when no view has *view_id*."""
+        from .models import utc_now_iso
+
+        trimmed = validate_name(name)
+        with self.model.get_connection() as conn:
+            conn.execute(
+                "UPDATE saved_views SET name = ?, updated_at = ? WHERE id = ?",
+                (trimmed, utc_now_iso(), view_id),
+            )
+            row = conn.execute(_SELECT + " WHERE id = ?", (view_id,)).fetchone()
+        # A hidden (unreadable) view is not found, as in the listing.
+        return _view(row) if row and _decode_object(row[3]) is not None else None
+
+    def delete_saved_view(self, view_id: int) -> bool:
+        """True when a view was removed."""
+        with self.model.get_connection() as conn:
+            cursor = conn.execute("DELETE FROM saved_views WHERE id = ?", (view_id,))
+        return cursor.rowcount > 0

database/vocabulary.py

@@ -0,0 +1,435 @@
+"""Bounded tag and category reads, and guarded tag commands.
+
+A count is the number of eligible Library parents that carry the value
+directly or through one active saved pair, which is exactly what
+``list_library(tags=[name])`` / ``list_library(category=value)`` return.
+
+Tag commands run in one ``_pair_transaction``: the storage epoch is checked
+first, then the exact ``{id, name}`` identity, and the receipt is read on the
+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.
+"""
+
+import sqlite3
+
+from .library import _ELIGIBLE, _limit
+from .library_commands import _TOKEN, LibraryCommandError, _validate_epoch
+from .library_tokens import canonical_token
+from .models import utc_now_iso
+
+try:
+    from ..utils.validators import validate_category, validate_tags
+except ImportError:
+    from utils.validators import validate_category, validate_tags
+
+_CATEGORY_DOMAIN = "prompt-manager-vocabulary-category:v1\n"
+
+
+def _tag_rows(page: str) -> str:
+    """Count rows for the registry rows selected by ``page`` (id, name)."""
+    # Eligible parent components carrying a tag in the page, plus the
+    # positive parent of every active pair carrying it (always eligible).
+    return (
+        "WITH page AS (" + page + "),"
+        "owners(tag_id,pid) AS ("
+        "SELECT pt.tag_id,p.id FROM prompts p "
+        "JOIN prompt_tags pt ON pt.prompt_id=p.id "
+        "WHERE pt.tag_id IN (SELECT id FROM page) AND " + _ELIGIBLE + " "
+        "UNION SELECT ppt.tag_id,pp.positive_prompt_id FROM prompt_pairs pp "
+        "JOIN prompt_pair_tags ppt ON ppt.pair_id=pp.id "
+        "WHERE pp.archived=0 AND ppt.tag_id IN (SELECT id FROM page)) "
+        "SELECT page.id,page.name,COUNT(DISTINCT owners.pid) AS prompt_count "
+        "FROM page LEFT JOIN owners ON owners.tag_id=page.id "
+        "GROUP BY page.id ORDER BY page.id"
+    )
+
+
+_TAG_PAGE = _tag_rows("SELECT id,name FROM tags WHERE id>? ORDER BY id LIMIT ?")
+_TAG_ONE = _tag_rows("SELECT id,name FROM tags WHERE id=?")
+# Every owner kind a tag can sit on, for the delete confirmation.
+_TAG_IMPACT = (
+    "SELECT "
+    "(SELECT COUNT(*) FROM prompts p JOIN prompt_tags pt ON pt.prompt_id=p.id "
+    "WHERE pt.tag_id=:tag AND " + _ELIGIBLE + ") AS positive_prompts,"
+    "(SELECT COUNT(*) FROM prompts p JOIN prompt_tags pt ON pt.prompt_id=p.id "
+    "WHERE pt.tag_id=:tag AND NOT " + _ELIGIBLE + ") AS other_prompts,"
+    "(SELECT COUNT(*) FROM prompt_pairs pp JOIN prompt_pair_tags ppt "
+    "ON ppt.pair_id=pp.id WHERE ppt.tag_id=:tag AND pp.archived=0) "
+    "AS negative_variants,"
+    "(SELECT COUNT(*) FROM prompt_pairs pp JOIN prompt_pair_tags ppt "
+    "ON ppt.pair_id=pp.id WHERE ppt.tag_id=:tag AND pp.archived!=0) "
+    "AS archived_negative_variants"
+)
+_MAX_ID = 2**63 - 1
+
+
+def validate_vocabulary_tag_name(value: object) -> str:
+    """Return the trimmed name, applying tag rules plus the native comma ban."""
+    if not isinstance(value, str):
+        raise ValueError("Tag name must be a string")
+    validate_tags([value])
+    name = value.strip()
+    if "," in name:
+        raise ValueError("Tag names cannot contain commas")
+    return name
+
+
+def _validate_tag_id(tag_id: object) -> None:
+    if type(tag_id) is not int or not 1 <= tag_id <= _MAX_ID:
+        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 = (
+    "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) "
+)
+
+
+# Every owner of one exact category value, in id order, with the flag the
+# impact counts split on.
+_CATEGORY_COMPONENTS = (
+    "SELECT p.id," + _ELIGIBLE + " AS eligible FROM prompts p "
+    "WHERE p.category=? ORDER BY p.id"
+)
+_CATEGORY_PAIRS = "SELECT id,archived FROM prompt_pairs WHERE category=? ORDER BY id"
+_CATEGORY_EXISTS = (
+    "SELECT EXISTS(SELECT 1 FROM prompts WHERE category=:v)"
+    " OR EXISTS(SELECT 1 FROM prompt_pairs WHERE category=:v)"
+)
+
+# Pairs after components, so a failing pair update rolls both back.
+_CATEGORY_UPDATES = (
+    "UPDATE prompts SET category=?,updated_at=? WHERE category=?",
+    "UPDATE prompt_pairs SET category=?,updated_at=? WHERE category=?",
+)
+
+
+def category_impact_token(
+    epoch: str,
+    source: str,
+    target: str,
+    target_exists: bool,
+    components: list,
+    pairs: list,
+) -> str:
+    """Hash a category projection like ``library_edit_token`` (no secret)."""
+    preimage = {
+        "epoch": epoch,
+        "source": source,
+        "target": target,
+        "target_exists": bool(target_exists),
+        "components": sorted([int(i), bool(flag)] for i, flag in components),
+        "pairs": sorted([int(i), bool(flag)] for i, flag in pairs),
+    }
+    return canonical_token(_CATEGORY_DOMAIN, preimage)
+
+
+def _validate_category_change(source: object, target: object) -> str:
+    """Check the exact source and return the trimmed, different target."""
+    if not isinstance(source, str) or not source.strip():
+        raise ValueError("source must be a nonblank category value")
+    if not isinstance(target, str):
+        raise ValueError("Category must be a string")
+    validate_category(target)
+    target = target.strip()
+    if not target:
+        raise ValueError("New category name cannot be blank")
+    if target == source:
+        raise ValueError("New category name must differ from the current one")
+    return target
+
+
+def _validate_impact_token(impact_token: object) -> None:
+    if not isinstance(impact_token, str) or not _TOKEN.fullmatch(impact_token):
+        raise ValueError("impact_token must be 64 lowercase hexadecimal characters")
+
+
+def _tag_after(after: object) -> int:
+    if after is None:
+        return 0
+    if type(after) is not int or after <= 0:
+        raise ValueError("Invalid vocabulary cursor")
+    return after
+
+
+def _category_after(after: object) -> tuple[int, str] | None:
+    if after is None:
+        return None
+    if (
+        not isinstance(after, list)
+        or len(after) != 2
+        or type(after[0]) is not int
+        or not isinstance(after[1], str)
+    ):
+        raise ValueError("Invalid vocabulary cursor")
+    return after[0], after[1]
+
+
+class VocabularyOperations:
+    """Vocabulary reads and tag commands.
+
+    ``_`` helpers take an open ``library_snapshot`` or ``_pair_transaction``
+    connection.
+    """
+
+    def _vocabulary_tag(self, conn: sqlite3.Connection, tag_id: int) -> dict | None:
+        row = conn.execute(_TAG_ONE, (tag_id,)).fetchone()
+        return dict(row) if row else None
+
+    def _vocabulary_identity(
+        self, conn: sqlite3.Connection, epoch: str, tag_id: int, name: str
+    ) -> dict:
+        """The stored tag, or vocabulary-conflict when it is gone or renamed."""
+        tag = self._vocabulary_tag(conn, tag_id)
+        if tag is None or tag["name"] != name:
+            raise LibraryCommandError("vocabulary-conflict", epoch)
+        return tag
+
+    @staticmethod
+    def _tag_owner_counts(conn: sqlite3.Connection, tag_id: int) -> dict:
+        components = conn.execute(
+            "SELECT COUNT(DISTINCT prompt_id) FROM prompt_tags WHERE tag_id=?",
+            (tag_id,),
+        ).fetchone()[0]
+        pairs = conn.execute(
+            "SELECT COUNT(DISTINCT pair_id) FROM prompt_pair_tags WHERE tag_id=?",
+            (tag_id,),
+        ).fetchone()[0]
+        return {"components_changed": components, "pairs_changed": pairs}
+
+    def create_vocabulary_tag(self, name: str, *, expected_epoch: str) -> dict:
+        """Insert an unused registry row, or reuse the exact existing name."""
+        _validate_epoch(expected_epoch)
+        name = validate_vocabulary_tag_name(name)
+        with self._pair_transaction() as conn:
+            epoch = self._library_epoch(conn, expected_epoch)
+            row = conn.execute("SELECT id FROM tags WHERE name=?", (name,)).fetchone()
+            created = row is None
+            if created:
+                tag_id = conn.execute(
+                    "INSERT INTO tags(name) VALUES(?)", (name,)
+                ).lastrowid
+            else:
+                tag_id = row[0]
+            return {
+                "library_epoch": epoch,
+                "created": created,
+                "tag": self._vocabulary_tag(conn, tag_id),
+            }
+
+    def rename_vocabulary_tag(
+        self, tag_id: int, *, expected_epoch: str, name: str, new_name: str
+    ) -> dict:
+        """Rename ``{tag_id, name}``, merging into an existing exact ``new_name``.
+
+        Links move across components and all pairs, archived included; an
+        owner already carrying the target keeps one link.
+        """
+        _validate_tag_id(tag_id)
+        _validate_epoch(expected_epoch)
+        if not isinstance(name, str):
+            raise ValueError("name must be the tag's current name")
+        new_name = validate_vocabulary_tag_name(new_name)
+        if new_name == name:
+            raise ValueError("New name must differ from the current name")
+        with self._pair_transaction() as conn:
+            epoch = self._library_epoch(conn, expected_epoch)
+            self._vocabulary_identity(conn, epoch, tag_id, name)
+            counts = self._tag_owner_counts(conn, tag_id)
+            target = conn.execute(
+                "SELECT id FROM tags WHERE name=?", (new_name,)
+            ).fetchone()
+            if target is None:
+                conn.execute("UPDATE tags SET name=? WHERE id=?", (new_name, tag_id))
+                result_id = tag_id
+            else:
+                result_id = target[0]
+                self._move_tag_links(conn, tag_id, result_id)
+                conn.execute("DELETE FROM tags WHERE id=?", (tag_id,))
+            return {
+                "library_epoch": epoch,
+                "merged": target is not None,
+                "tag": self._vocabulary_tag(conn, result_id),
+                **counts,
+            }
+
+    def delete_vocabulary_tag(
+        self, tag_id: int, *, expected_epoch: str, name: str
+    ) -> dict:
+        """Remove the registry row ``{tag_id, name}``; its links cascade."""
+        _validate_tag_id(tag_id)
+        _validate_epoch(expected_epoch)
+        if not isinstance(name, str):
+            raise ValueError("name must be the tag's current name")
+        with self._pair_transaction() as conn:
+            epoch = self._library_epoch(conn, expected_epoch)
+            self._vocabulary_identity(conn, epoch, tag_id, name)
+            counts = self._tag_owner_counts(conn, tag_id)
+            conn.execute("DELETE FROM tags WHERE id=?", (tag_id,))
+            return {"library_epoch": epoch, "deleted": True, **counts}
+
+    def vocabulary_tag_impact(self, tag_id: int) -> dict | None:
+        """Owner counts by kind for one tag, or None when it does not exist."""
+        _validate_tag_id(tag_id)
+        with self.library_snapshot() as (conn, epoch, _ceiling):
+            if not conn.execute("SELECT 1 FROM tags WHERE id=?", (tag_id,)).fetchone():
+                return None
+            row = conn.execute(_TAG_IMPACT, {"tag": tag_id}).fetchone()
+            return {"library_epoch": epoch, "impact": dict(row)}
+
+    @staticmethod
+    def _category_projection(
+        conn: sqlite3.Connection, epoch: str, source: str, target: str
+    ) -> dict:
+        """Owners of ``source``, impact counts and the token over them."""
+        components = [
+            [row[0], bool(row[1])]
+            for row in conn.execute(_CATEGORY_COMPONENTS, (source,))
+        ]
+        pairs = [
+            [row[0], bool(row[1])] for row in conn.execute(_CATEGORY_PAIRS, (source,))
+        ]
+        if not components and not pairs:
+            raise LibraryCommandError("vocabulary-conflict", epoch)
+        target_exists = bool(
+            conn.execute(_CATEGORY_EXISTS, {"v": target}).fetchone()[0]
+        )
+        eligible = sum(flag for _id, flag in components)
+        archived = sum(flag for _id, flag in pairs)
+        return {
+            "target_exists": target_exists,
+            "impact": {
+                "positive_prompts": eligible,
+                "other_prompts": len(components) - eligible,
+                "negative_variants": len(pairs) - archived,
+                "archived_negative_variants": archived,
+            },
+            "impact_token": category_impact_token(
+                epoch, source, target, target_exists, components, pairs
+            ),
+        }
+
+    def preview_vocabulary_category(
+        self, *, expected_epoch: str, source: str, target: str
+    ) -> dict:
+        """Read-only impact of renaming the exact ``source`` to ``target``."""
+        _validate_epoch(expected_epoch)
+        target = _validate_category_change(source, target)
+        with self.library_snapshot() as (conn, epoch, _ceiling):
+            if epoch != expected_epoch:
+                raise LibraryCommandError("storage-changed", epoch)
+            projection = self._category_projection(conn, epoch, source, target)
+            return {
+                "library_epoch": epoch,
+                "source": source,
+                "target": target,
+                **projection,
+            }
+
+    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.
+
+        The recomputed projection must still match the preview's token; a
+        mismatch is ``vocabulary-conflict`` with nothing written.
+        """
+        _validate_epoch(expected_epoch)
+        target = _validate_category_change(source, target)
+        _validate_impact_token(impact_token)
+        with self._pair_transaction() as conn:
+            epoch = self._library_epoch(conn, expected_epoch)
+            projection = self._category_projection(conn, epoch, source, target)
+            if projection["impact_token"] != impact_token:
+                raise LibraryCommandError("vocabulary-conflict", epoch)
+            stamp = utc_now_iso()
+            for statement in _CATEGORY_UPDATES:
+                conn.execute(statement, (target, stamp, source))
+            return {
+                "library_epoch": epoch,
+                "source": source,
+                "target": target,
+                "consolidated": projection["target_exists"],
+                "impact": projection["impact"],
+            }
+
+    def _vocabulary_tags(
+        self,
+        conn: sqlite3.Connection,
+        epoch: str,
+        *,
+        limit: int = 100,
+        after: int | None = None,
+    ) -> dict:
+        _limit(limit)
+        start = _tag_after(after)
+        total = conn.execute("SELECT COUNT(*) FROM tags").fetchone()[0]
+        rows = [dict(row) for row in conn.execute(_TAG_PAGE, (start, limit + 1))]
+        has_more = len(rows) > limit
+        rows = rows[:limit]
+        return {
+            "library_epoch": epoch,
+            "tags": rows,
+            "total": total,
+            "has_more": has_more,
+            "next_key": rows[-1]["id"] if has_more else None,
+        }
+
+    def _vocabulary_categories(
+        self,
+        conn: sqlite3.Connection,
+        epoch: str,
+        *,
+        limit: int = 100,
+        after: list | None = None,
+    ) -> dict:
+        _limit(limit)
+        key = _category_after(after)
+        total = conn.execute(
+            _CATEGORY_GROUPS + "SELECT COUNT(*) FROM grouped"
+        ).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"
+            + where
+            + " ORDER BY first_pid,value LIMIT ?",
+            [*params, limit + 1],
+        ).fetchall()
+        has_more = len(rows) > limit
+        rows = rows[:limit]
+        return {
+            "library_epoch": epoch,
+            "categories": [
+                {"value": row["value"], "prompt_count": row["prompt_count"]}
+                for row in rows
+            ],
+            "total": total,
+            "has_more": has_more,
+            "next_key": (
+                [rows[-1]["first_pid"], rows[-1]["value"]] if has_more else None
+            ),
+        }
+
+    def list_vocabulary_tags(self, *, limit: int = 100, after: int | None = None):
+        with self.library_snapshot() as (conn, epoch, _ceiling):
+            return self._vocabulary_tags(conn, epoch, limit=limit, after=after)
+
+    def list_vocabulary_categories(
+        self, *, limit: int = 100, after: list | None = None
+    ) -> dict:
+        with self.library_snapshot() as (conn, epoch, _ceiling):
+            return self._vocabulary_categories(conn, epoch, limit=limit, after=after)

docs/application-redesign.md

@@ -0,0 +1,84 @@
+# Application Redesign Development Plan
+
+The maintainer approved the [interactive concept](concepts/workspace/README.md)
+on 2026-10-17. It is a design reference with simulated operations, not production
+code. This plan proposes implementation order and acceptance gates.
+
+## Delivery status
+
+- Foundation merged in [#198](https://github.com/ComfyAssets/ComfyUI_PromptManager/pull/198).
+- [#200](https://github.com/ComfyAssets/ComfyUI_PromptManager/issues/200) delivers native Images browsing, physical-file details, infinite scroll, Grid/List/Masonry, Viewer.js and video playback. The workspace opens on Images; original application entry points remain available.
+- Shared role-aware sending and the physical-file prompt inspector merged in [#203](https://github.com/ComfyAssets/ComfyUI_PromptManager/pull/203) and [#206](https://github.com/ComfyAssets/ComfyUI_PromptManager/pull/206).
+- The native Prompt Library merged in [#210](https://github.com/ComfyAssets/ComfyUI_PromptManager/pull/210) for [#208](https://github.com/ComfyAssets/ComfyUI_PromptManager/issues/208): one positive parent, saved negative variants, full selected text and a shared filtered image gallery. It does not invent chronological runs from counters or scan dates.
+- [Durable history storage](prompt-history-storage.md) records job attempts and outcomes from [#212](https://github.com/ComfyAssets/ComfyUI_PromptManager/pull/212), frozen prompt contexts from [#213](https://github.com/ComfyAssets/ComfyUI_PromptManager/pull/213), and observed run-owned output locators from [#214](https://github.com/ComfyAssets/ComfyUI_PromptManager/pull/214). Those locators do not verify file bytes or establish gallery-image associations. That association contract and the chronological execution timeline remain pending under [#209](https://github.com/ComfyAssets/ComfyUI_PromptManager/issues/209).
+- [#215](https://github.com/ComfyAssets/ComfyUI_PromptManager/issues/215) brings the shell, Images and natural-height Prompt Library presentation into the approved geometry and themes while retaining their existing operations. [#216](https://github.com/ComfyAssets/ComfyUI_PromptManager/issues/216) adds the shared current-file inspector with exact generation facts, scoped provenance, raw/download evidence and loaded-collection viewer controls, including a slideshow that pauses at native video. This delivery does not establish whole-application parity, change the default entry point or constitute a release. Recorded chronological run/context browsing continues under [#209](https://github.com/ComfyAssets/ComfyUI_PromptManager/issues/209); image tagging/review, System consolidation and remaining organization capabilities continue under #196.
+
+## Scope and compatibility
+
+Rebuild the standalone application around one shared shell. Preserve existing
+node schemas, saved workflows, backend capabilities, pinned Viewer.js and the
+ComfyAssets toolkit. Keep existing application entry points usable during the
+transition. Do not change the running maintainer instance during automated tests.
+
+Use the concept's feature inventory as a parity checklist. Each row needs an
+implementation issue, observable acceptance criteria and automated test coverage.
+Record any intentional behavior change explicitly; no silent feature removal.
+
+## Proposed delivery sequence
+
+Use a tracking issue and small issue-linked branches and PRs. Dependencies may
+require splitting these milestones further.
+
+| Milestone | Deliverable and acceptance focus |
+| --- | --- |
+| 1. Foundation | Production shell, routing, shared theme and disposable page-test harness. Verify startup, navigation, dark/light changes, cleanup and existing entry points. Establish discovered-test and coverage baselines before changes. |
+| 2. Image workspace | Real API-backed filters, Grid/List/Masonry, readable thumbnails, selection and shared Viewer.js inspector. Continuous infinite scroll only, with accessible Load more fallback, stable ordering, retry and restored browsing context. Bound retained DOM and test large datasets. |
+| 3. Prompt library and history | Real positive/negative associations, selected full negative, run history and shared image filtering. Design any missing execution-history persistence and migration before wiring the timeline; never invent historical runs from incomplete metadata. |
+| 4. Send to ComfyUI | Shared paper-plane actions wherever prompt text appears; role-specific and pair sends, destination preview, acknowledgement and failure recovery. Cover explicit empty versus unknown negatives and connected inputs. Never queue or replace a workflow. |
+| 5. Editing and review | Prompt editing, vocabulary management, per-image manual/auto-tag approval and Needs review. Preserve cancellation semantics and distinguish tag approval from prompt-association decisions. |
+| 6. System and maintenance | Settings, sources/scans, integrations, jobs, diagnostics, backup/restore and maintenance impact previews. Cover existing capabilities before retiring their old controls. |
+| 7. Additional organization | Favorites independent of ratings, collections, saved views, Activity, scheduled backups and portable import previews. Specify persistence, validation and recovery separately for each capability. |
+| 8. Cutover | Complete the parity checklist, migration and upgrade validation, accessibility/performance checks, documentation and targeted maintainer live testing. Switch the default only after these pass; preserve compatible URLs. |
+
+## TDD and test layers
+
+For each behavior, first write and run a meaningful failing test. Implement the
+smallest working change, verify it passes, then refactor with the suite green.
+Include failing-before/passing-after evidence in the PR.
+
+- **Unit:** filtering and selection state, cursor handling, prompt roles,
+  formatting, validation, navigation and lifecycle cleanup.
+- **Integration:** real HTTP handlers and disposable SQLite/filesystem fixtures;
+  persistence, migration from supported old schemas, idempotent scans, restore,
+  rollback and reference-safe deletion. Mock inference and external services.
+- **Smoke:** shipped production pages start and navigate with real fixture APIs;
+  test populated, empty and error states, assets and unexpected console failures.
+- **Browser workflows:** production HTML/JS across Chromium, Firefox, WebKit and
+  touch; editing/reload, infinite append, viewer cleanup, tagging approval,
+  keyboard access, delayed responses, cancellation and retries.
+- **Host integration:** pinned disposable CPU ComfyUI, Classic and Nodes 2.0,
+  switching both ways and workflow reload for host-facing changes. Exercise
+  independent and combined pack installations for shared bridge/toolkit changes.
+
+Retain existing test discovery and coverage gates. Carry forward database
+injection and open-handle checks into new fixtures. Add new frontend modules to
+coverage measurement and its baseline; do not lower floors or replace existing
+regression coverage with mockup-only tests.
+
+## GitHub and release gates
+
+Extend existing required workflows: the Python 3.10–3.13 matrix on Linux,
+Windows and macOS; 85% Python coverage; formatting; frontend units and per-file
+coverage floors; vendor checks; browser workflows; real ComfyUI compatibility.
+Publish traces, screenshots, logs and coverage artifacts on failures. See
+[TESTING.md](../TESTING.md) for commands and fixture isolation requirements.
+
+Every PR must map acceptance criteria to tests, report relevant checks and include
+screenshots for visual changes. All required checks must pass before merging.
+Use maintainer live review at meaningful visual and host-integration milestones;
+capture discovered regressions in automated tests. Notify the maintainer when
+Python changes require a ComfyUI restart.
+
+Keep version/publication changes separate from intermediate UI work: the existing
+registry workflow publishes on main/master changes to `pyproject.toml` or manual
+dispatch. Do not trigger a release merely to preview a redesign milestone.

docs/concepts/workspace/README.md

@@ -0,0 +1,138 @@
+# Prompt Manager application redesign concept
+
+Status: visual concept approved by the maintainer, 2026-10-17. Production
+implementation is progressing in tested slices; see the [development plan](../../application-redesign.md).
+The production workspace now has native Images browsing; this concept remains the
+approved target for the complete application.
+
+## Review the mockup
+
+Run `python tools/preview-prompt-timeline.py --bind 0.0.0.0 --port 8767` and
+open `/workspace/`. The original approved timeline remains available at `/`.
+The server exposes an explicit allowlist of concept files and pinned assets.
+All examples are synthetic; there are no application API calls, database access,
+real model downloads, or ComfyUI handoffs. Local file inspection previews an
+image in the browser without uploading it. Exports contain sample data only.
+
+The sidebar opens Images, Prompt library, Tags & categories, Inspect a file,
+Activity, System and Design review. System includes Overview, Appearance,
+Sources & scans, Integrations, Auto-tagging, Maintenance, Backup & export, and
+Diagnostics & logs. The in-app Design review contains the feature map below.
+
+## Proposed direction
+
+One application shell replaces separate Admin, Gallery and Metadata shells.
+Use the approved shared ComfyAssets theme: ComfyUI dark/light, blue primary
+actions, green positive roles, red negative roles and errors. Preserve existing
+capabilities and services while consolidating controls and clarifying their scope.
+
+Browsing is image-led with readable thumbnails and one metadata inspector.
+Prompt library groups by positive; negative variants and runs share its gallery.
+The full selected negative appears directly beneath the positive in history.
+Favorites express preference independently of the existing quality rating.
+Collections organize projects without duplicating recipe text; saved views keep
+search/filter context. New capabilities are proposals, not implemented features.
+
+Maintenance moves out of the browsing toolbar. Checks, impact preview, backups,
+and apply steps have separate responsibilities. Distinguish prompt text reuse
+from image-file duplicates, and metadata enrichment from thumbnail generation.
+No repeated global thumbnail-generation prompt on opening the gallery.
+
+## Send to ComfyUI
+
+Use the same original paper-plane outline, blue accent and accessible role label
+on every positive/negative prompt block, including image details and the viewer.
+A pair action sends both. Preview destination and exact fields before applying;
+positive-only and negative-only actions leave the other role untouched. An
+explicitly empty negative means clear the target field; unknown metadata must
+never be interpreted as empty. Sending text neither queues execution nor
+replaces a workflow.
+
+Production needs a shared host bridge with versioned payloads, origin validation,
+destination discovery, capability checks, disconnected-host handling, linked-input
+checks and acknowledgements. Preview actual target values and report success only
+after acknowledgement. Default to asking for a destination; never guess between
+multiple nodes or positive and negative encoder inputs. Verify Classic and Nodes
+2.0 with both packs installed independently and together.
+
+## Existing feature inventory
+
+| Current feature | Proposed home / behavior | Disposition |
+| --- | --- | --- |
+| Admin prompt cards / per-prompt gallery | Prompt library + one prompt history workspace | Retained, rebuilt |
+| Standalone image gallery / Admin image modal | One Images workspace + shared Viewer.js viewer | Consolidated |
+| Video playback, autoplay / mute / loop, viewer slideshow and transforms | Shared media viewer; playback controls in Appearance; same image navigation | Retained |
+| Metadata page / viewer sidebar / full-text popups | One reusable metadata inspector; inline complete prompt text | Consolidated |
+| Prompt add, edit, rating, notes, copy, delete | Recipe editor, shared details, contextual bulk actions; reference-safe archive | Retained, clarified |
+| Categories, tag rename / merge, autocomplete | Tags & categories; same vocabulary in every editor | Retained |
+| Grid/list, thumbnail size, columns, pages/infinite scroll | Infinite scroll only; one Appearance configuration and contextual layout / density overrides | Simplified |
+| Scan, rescan, monitoring directories, worker threads | Sources & scans; explicit metadata enrichment vs thumbnail work | Consolidated |
+| Ollama, HuggingFace, WD14 variants, thresholds, review | One Auto-tagging configuration and review flow | Retained |
+| Thumbnail quality, size limit, metadata cache, paths | Appearance + cache controls in Maintenance | Retained |
+| Prompt dedup, hash checks, consistency, unused prompts | Maintenance plan with dependency checks and impact preview | Retained, safer flow |
+| Image dedup, orphan image links, browser cache | Separate reviewed operations in Maintenance | Retained |
+| Database backup / restore, JSON export | Backup & export; recovery and portable recipes clearly distinguished | Retained |
+| Diagnostics, image-link checks, logs, rotation, log levels | System → Diagnostics & logs | Retained |
+| LoRA detection/import, trigger injection, CivitAI key | System → Integrations; positive-only triggers | Retained |
+| Statistics / counts | System overview + usage breakdown; context counts in the workspace | Retained |
+| Send prompts to ComfyUI from every prompt surface | Role-aware paper-plane action + destination / overwrite preview | New proposal |
+| Reusable negatives / approved branch timeline | Prompt history with selected negative directly beneath positive | Approved concept |
+| Favorites, saved searches, project collections | Sidebar views; collections do not duplicate recipe text | New proposal |
+| Ambiguous metadata review, source provenance, run comparisons | Needs review queue; recorded vs inferred evidence; pair/run selection | New / expanded |
+| Unified jobs, cancellation, resumable progress visibility | Activity center; cancellation distinguished from hiding progress | New proposal |
+| Scheduled backups, pre-change safety snapshots | Explicit recovery policy after storage configuration | New proposal |
+
+## Implementation boundaries and test requirements
+
+Keep Viewer.js 1.15.2 and the shared toolkit as reproducible pinned artifacts;
+this mock uses their actual checked-in files without modifying them. Preserve
+legacy routes during the transition, current node layouts/schemas, existing
+controllers where reusable, and all existing test/coverage gates.
+
+The timeline needs persisted execution records to represent real runs. Saved
+views, collections, metadata review resolution, activity history, scheduled
+backups, portable JSON import, and the ComfyUI bridge need separate implementation
+and persistence work. Preserve source evidence and distinguish recorded from
+inferred metadata. Existing schema migration is automatic; old image enrichment
+remains an optional explicit scan. The one-time upgrade notice belongs in the
+new shell, with Scan and Later choices.
+
+Implement in reviewed slices: shared application shell and inspector; reusable
+prompt blocks/send bridge; image workspace; grouped library/history; settings and
+maintenance; optional organization and workflow improvements. Write regressions
+first for production behavior. Retain Python, frontend unit, Chromium/Firefox/
+WebKit/touch and managed Classic/Nodes 2.0 tests. Add behavior tests for shared
+filters, navigation state, viewer cleanup, interrupted jobs, dependency-safe
+maintenance, backup/restore, paired metadata and correct send destinations.
+
+Screenshots in `screenshots/` record the reviewed starting point. The approved
+standalone timeline in `../prompt-variant-timeline.html` remains unchanged.
+
+## Review polish
+
+History actions use a shared branch icon. Images can switch among Grid, List and
+Masonry; Masonry preserves each image's proportions and keeps DOM order while
+packing cards with a responsive layout. It does not require experimental native
+masonry support. Resize observers are disconnected when leaving the view.
+
+Every image card includes a tagging action beside Info, including cards in Needs
+review. Choose WD14 SwinV2, WD14 ViT, Ollama or HuggingFace; adjust thresholds or
+instructions, generate sample suggestions, edit tags manually, and approve only
+the desired suggestions. Changes are staged until approval, apply to that image
+alone and do not resolve uncertain prompt associations. Changing models discards
+previous suggestions. Cancel preserves existing tags. All inference is simulated.
+
+## Continuous browsing decision
+
+The redesign uses infinite scroll only. Remove page numbers, previous/next page
+buttons, paged/infinite mode selection and results-per-page controls. Keep request
+batching internal. The mock appends 12 images at a time in Grid, List and Masonry;
+it preserves existing card nodes, focus and selection while appending. It also
+retains the loaded range when switching layouts or revisiting a search. A
+keyboard-accessible Load more action provides an alternate trigger for the same
+continuous stream, not pagination. Exhausted and empty results are explicit.
+
+Production will need bounded rendering, stable cursors, scroll restoration and
+loading/error/retry behavior for large collections. Existing backend pagination
+can continue providing internal request batches; no production tests or routes
+are removed as part of this prototype.

docs/concepts/workspace/index.html

@@ -0,0 +1,105 @@
+<!doctype html>
+<html lang="en">
+  <head>
+    <meta charset="utf-8" />
+    <meta name="viewport" content="width=device-width,initial-scale=1" />
+    <meta name="referrer" content="no-referrer" />
+    <title>Images · Prompt Manager concept</title>
+    <link rel="stylesheet" href="/toolkit/controls.css" />
+    <link rel="stylesheet" href="/vendor/viewer.css" />
+    <link rel="stylesheet" href="/workspace/style.css" />
+    <script src="/vendor/viewer.js" defer></script>
+    <script src="/workspace/app.js" type="module"></script>
+  </head>
+  <body class="ca-tools">
+    <a class="skip" href="#main">Skip to workspace</a>
+    <div class="app-shell">
+      <aside class="sidebar">
+        <a class="brand" href="#images"
+          ><span class="brand-mark">P<span>±</span></span
+          ><span>Prompt Manager<small>COMFYASSETS</small></span></a
+        >
+        <div class="sidebar-label">WORKSPACE</div>
+        <nav id="primary-nav" aria-label="Workspace"></nav>
+        <div class="sidebar-label">
+          SAVED VIEWS <span class="new-label">PROPOSED</span>
+        </div>
+        <nav id="saved-nav" aria-label="Saved views"></nav>
+        <div class="sidebar-label">
+          COLLECTIONS <span class="new-label">PROPOSED</span>
+        </div>
+        <nav id="collections-nav" aria-label="Collections"></nav>
+        <div class="sidebar-bottom">
+          <nav id="system-nav" aria-label="Tools and system"></nav>
+          <div class="connection">
+            <i></i>
+            <div>
+              ComfyUI workspace<small
+                >Demo connection · Classic / Nodes 2.0</small
+              >
+            </div>
+          </div>
+        </div>
+      </aside>
+      <div class="workspace">
+        <header class="topbar">
+          <div class="breadcrumbs">
+            ComfyAssets <span>/</span> <span id="breadcrumb">Images</span>
+          </div>
+          <div class="top-actions">
+            <span class="concept-badge">Interactive concept</span
+            ><button id="activity-open" type="button">
+              Activity <span class="count" id="activity-count">2</span></button
+            ><label class="theme-label" for="theme"
+              >Theme
+              <select id="theme" aria-label="Theme">
+                <option value="dark">Dark</option>
+                <option value="light">Light</option>
+                <option value="system">System</option>
+              </select></label
+            ><a class="canvas-link" href="#send"
+              >ComfyUI <span aria-hidden="true">↗</span></a
+            >
+          </div>
+        </header>
+        <main id="main" tabindex="-1"></main>
+        <footer class="app-footer">
+          <span
+            >Design prototype · sample images and data · actions stay in this
+            tab</span
+          ><button type="button" id="review-link">What changes & why</button>
+        </footer>
+      </div>
+    </div>
+    <div id="toast" class="toast" role="status" aria-atomic="true"></div>
+    <dialog id="modal" aria-labelledby="modal-title">
+      <div class="dialog-header">
+        <h2 id="modal-title"></h2>
+        <button type="button" id="modal-close" aria-label="Close dialog">
+          ×
+        </button>
+      </div>
+      <div id="modal-body"></div>
+    </dialog>
+    <dialog id="viewer-dialog" aria-labelledby="viewer-title">
+      <div class="dialog-header">
+        <div>
+          <h2 id="viewer-title">Image details</h2>
+          <p class="muted small">
+            One viewer · image, prompts, generation data, workflow
+          </p>
+        </div>
+        <button type="button" id="viewer-close" aria-label="Close image viewer">
+          ×
+        </button>
+      </div>
+      <div class="viewer-layout">
+        <div class="viewer-left">
+          <div id="viewer-stage"></div>
+          <div class="viewer-toolbar" id="viewer-toolbar"></div>
+        </div>
+        <aside id="viewer-details" aria-label="Image metadata"></aside>
+      </div>
+    </dialog>
+  </body>
+</html>

docs/prompt-history-storage.md

@@ -0,0 +1,84 @@
+# Durable Prompt History
+
+Status: the durable job ledger, immutable prompt contexts and run-owned output
+references are implemented for
+[#209](https://github.com/ComfyAssets/ComfyUI_PromptManager/issues/209). Gallery
+image associations and the chronological timeline remain pending; #209 stays open.
+
+## Recorded jobs
+
+New jobs observed at the ComfyUI queue's dequeue boundary receive an immutable
+run identity and a host-session/queue-item identity. The public ComfyUI job UUID
+is a correlation field: submitting it twice records two attempts. Rejected graphs
+and jobs removed from the queue before dequeue create no execution rows. Cached
+jobs are recorded even when no PromptManager node runs, and headless jobs do not
+require a browser or WebSocket client.
+
+The additive `execution_runs` table stores dequeue time, observed callback time,
+optional executor event times and an explicit outcome: `observed`, `succeeded`,
+`failed`, `interrupted` or `unknown`. Matching stored terminal events determine
+outcomes. Missing or conflicting terminal evidence remains unknown; a process
+crash can leave an observed row without an invented completion time. Bounded
+safe diagnostics retain node ID/type and exception type, never graph text,
+arbitrary extra data, credentials, exception messages, traceback, inputs or
+arbitrary output payloads. A bounded background writer keeps storage failures from
+changing host execution and reports capture gaps rather than inventing rows.
+
+Existing databases begin with zero execution rows. Migration and backup restore
+preserve prompts, pairs, image links and aggregate counters; storage epochs keep
+stale queued writes from being replayed into restored data. There is no
+historical backfill from counters, scan dates, image timestamps or saved recipes.
+
+[Testing](../TESTING.md#durable-execution-ledger) describes the real CPU host
+matrix, historical failure proof, privacy checks and cleanup. Python deployment
+requires restarting ComfyUI.
+
+## What the existing UI proves
+
+`prompts.run_count` and `prompt_pairs.run_count` remain aggregate usage counters,
+not completed-run counts. `last_used_at` is the latest aggregate update; legacy
+migration may have estimated counts from linked images. `generated_images`
+identifies linked files; image counts and insertion times are not run counts or
+execution times. Image/pair association creation time can reflect an old-image
+scan. A saved pair's creation time is its recipe date.
+
+The native Prompt Library shows saved negative variants, linked images and
+aggregate uses. It does not expose the ledger as chronological dots or repeat-run
+branches. `PromptTracker` identities and PNG recipe evidence remain separate
+from durable host job identity.
+
+## Immutable contexts and output references
+
+Prompt contexts retain bounded raw/effective positive, negative and composition
+snapshots with nullable recipe navigation hints and explicit evidence provenance.
+Unknown text remains SQL NULL; a known empty negative remains empty text. Cached
+evidence belongs to its exact cache provider and attempt, rather than inheriting
+unrelated recent prompts. Deletion and restore can remove navigation hints without
+rewriting historical snapshots or rebinding them to current library rows.
+
+Schema v3 adds `execution_output_capture` and `execution_run_outputs` without
+backfilling older runs or modifying gallery image links. At final queue completion,
+the writer attaches bounded output evidence from that attempt's final host history
+to its immutable run ID. The internal `get_execution_outputs(run_id)` reader
+returns state, reason, format version, source and ordered references. A complete
+capture with `none_reported` means the final host history reported no images;
+`not_captured`, partial, unavailable and conflicting evidence remain explicit.
+Failed and interrupted jobs can still have complete evidence for images reported
+before termination. Running jobs do not gain a final capture merely because an
+image node finished.
+
+Each reference records output node ID, image ordinal, folder type, subfolder,
+filename, path style and an opaque configured-root key. A cached repeat can retain
+the same locators under a new run. Locators do not verify file bytes, continued
+existence, a newly generated file or membership in any prompt context: the reader
+reports file identity as `unverified`, generation and context membership as
+`unknown`, and temp outputs as transient. Capture never reads, resolves or serves
+the files and adds no public route. PNG metadata is not required. Same-path file
+replacement and existing gallery provenance retain their separate semantics.
+
+## Remaining history work
+
+Define a deduplicated run/gallery-image relationship that preserves existing image
+provenance and same-path replacement protection. Add a chronological timeline only
+after its navigation and evidence semantics can honestly represent prompt-level
+execution chronology, batched outputs and repeat-run branches.

docs/prompt-inspector.md

@@ -0,0 +1,34 @@
+# Prompt inspection and sending
+
+The opt-in workspace at `/prompt_manager/workspace#/images` shows full positive
+and negative prompts in both Image details and the full-image preview. Existing
+gallery and metadata tools remain available during the application rebuild.
+
+Each prompt shows its evidence source: recorded execution, workflow inference,
+or embedded generation parameters. Path-only saved-library associations are
+marked unverified and cannot be sent: a filename alone cannot prove that an image
+has not been replaced. Missing, ambiguous, and unreadable prompts remain visible
+as distinct states. An explicitly
+empty negative is a resolved value and can clear a destination's negative field.
+
+Use the blue paper-plane action beside a prompt, or **Send prompt pair**. Keep a
+ComfyUI tab open under the same origin and deployment prefix. Choose a window and
+compatible Prompt Manager node, select **Preview changes**, review the complete
+before/after text and preserved prepend/append composition, then **Apply to
+ComfyUI**. The dialog reports success only after a matching acknowledgement.
+It does not load workflows or queue generation. If confirmation is lost, inspect
+ComfyUI and preview again before another attempt. Closing the dialog cannot undo
+an update already sent.
+
+## Read-only API
+
+`GET /prompt_manager/images/inspect/{relative_path}?root={root_index}` requires
+an explicit configured root, including when two roots contain the same filename.
+The response contains `image` identity and `prompts.positive` / `prompts.negative`.
+Each role has `status` (`resolved`, `absent`, `unresolved`, or `ambiguous`), `text`
+(string only when resolved, otherwise null), and `source`. Inspection does not
+index files, alter saved prompts, run scans, or decode image pixels. PNG text and
+video probe output are bounded; unsupported or malformed evidence remains
+unresolved rather than becoming sendable placeholder text.
+
+Python route changes require restarting ComfyUI. Browser assets require a refresh.

docs/prompt-send-bridge.md

@@ -0,0 +1,49 @@
+# Prompt sending bridge
+
+Issue #202 establishes the transport and host adapter before the redesigned
+inspector displays prompt text. Gallery paper-plane controls and normalized
+metadata follow in a separate slice; existing node layouts are unchanged.
+
+## Shared transport
+
+PromptManager consumes the exact Toolkit 0.2.0 artifact recorded in
+`web/js/shared/toolkit/provenance.json`. The shared module has no ComfyUI imports.
+Its client discovers explicit host sessions and compatible destinations, requests
+a preview, and applies that preview using a one-use, client-bound, expiring ticket.
+Success requires an acknowledgement with verified destination values.
+
+BroadcastChannel connects independently opened tabs on the same origin and storage
+partition. Deployment prefixes have separate channel namespaces. Exact origin,
+protocol and peer checks reject unrelated messages. Same-origin scripts are
+trusted; this transport is not authentication against other scripts on that origin.
+Unsupported messaging is reported without breaking existing ComfyUI nodes.
+
+An apply timeout or abort after posting means the outcome is unknown. It cannot
+prove the host cancelled the write. The sender must inspect a fresh destination
+preview before deciding to send again; automatic retry must not overwrite edits.
+
+## PromptManager adapter
+
+Classic, Text, Focused and Split accept positive text; +/- accepts either role or
+both. Plain CLIP encoders are not assigned roles by title or position. An omitted
+role remains untouched; an explicitly empty string clears only the selected
+field. Unknown/ambiguous metadata must not become an empty string.
+
+Previews show original and requested text plus positive prepend/append composition.
+Connected composition is marked unknown rather than guessing upstream values.
+Connected destination fields reject writes. Applying rechecks graph, node, widgets,
+values and connections. Ordinary serialization cache synchronization is accepted
+only when it preserves live values. Both requested roles are checked before writing.
+
+Widget values, visible inputs, serialization and composed previews are synchronized.
+Callback failures restore local fields; arbitrary external callback side effects
+cannot be undone. Sending never queues execution or loads/replaces a workflow.
+
+## Lifecycle and validation
+
+Page hiding closes the host and invalidates tickets; restored pages mount one fresh
+host. Client disposal clears pending timers/listeners and closes its channel.
+Shared protocol contracts run against the actual bundled file, alongside adapter
+units, cross-browser tab workflows and real Classic/Nodes 2.0 scenarios. Pinned
+PromptManager-only, KikoTools-only and combined disposable hosts verify coexistence
+and workflow reload without modifying either live installation.

docs/reusable-prompt-pairs.md

@@ -0,0 +1,111 @@
+# Reusable positive and negative prompts
+
+Tracking: [#193](https://github.com/ComfyAssets/ComfyUI_PromptManager/issues/193).
+This work builds on the merged Split node in PR #192. Keep the feature unmerged
+until maintainer live testing is complete.
+
+## Storage contract
+
+Keep existing prompt IDs, text hashes, metadata, image links and node schemas.
+A component can serve either role. Ordered combinations refer to shared prompt
+rows; an explicitly empty negative uses a null reference. Missing or ambiguous
+metadata does not mean an empty negative. Pair metadata belongs to the
+combination; saving another combination does not overwrite component metadata.
+Existing trim/lowercase hash normalization remains unchanged.
+
+Saved components are immutable while referenced. To change text, create or reuse
+a component and save a new combination. Archive combinations that have generation
+history. Pruning and legacy editing/deletion must respect these dependencies.
+Historical workflow snapshots remain intact, including their original text.
+
+## Migration and recovery
+
+Create pair tables after existing legacy rebuild migrations, in an atomic,
+repeatable migration. Never scan media during schema initialization. Test fresh,
+legacy and current databases, repeated startup, injected failures and retries.
+Preserve old backup compatibility. Migrate and validate a disposable backup copy
+before replacing live database pages. Reject broken pair schemas, foreign keys,
+triggers and views. Use SQLite backup APIs so committed WAL data is included.
+
+## Metadata and scanning
+
+Resolve positive and negative from the same sampler/output context. Track both
+node ID and output slot; bound graph traversal and text construction. Unknown
+conditioning operations and conflicting sampler candidates remain unresolved.
+Never guess that the first encoder or most recent negative belongs to an image.
+
+Import high-confidence pairs atomically with their image association and source
+provenance. Repeated scans must not duplicate components, pairs or image links,
+regenerate thumbnails, or increment generation run counts. Preserve existing
+conflicting/manual associations for review. Report new, enriched, unchanged,
+ambiguous, unresolved and failed results separately.
+
+## Live node
+
+Use the shared toolkit and Split foundation. Positive and negative editors have
+equal size. Only positive receives prepend, append and automatic LoRA triggers.
+Keep upstream connections authoritative. Library selection previews both fields
+before an explicit load. Saving text changes creates another reusable combination.
+Verify Classic and Nodes 2.0 startup, switching both directions, reload, caching,
+linked inputs, popup behavior and lifecycle cleanup before maintainer testing.
+
+## Delivery gate
+
+Keep all existing coverage floors and test suites. Record regression failures
+before implementing each behavior. Tests use disposable state and the managed
+CPU host; never mutate the maintainer's running ComfyUI or personal database.
+Python changes require a ComfyUI restart. Live approval is required before merge.
+
+## Live testing
+
+1. Download a database backup from Admin before the first restart. The additive
+   schema migration runs on startup; existing workflows and prompt IDs stay valid.
+2. Restart ComfyUI, reload the browser, and add **Prompt Manager (+/-)** from
+   **ComfyAssets / Prompts**. Connect CLIP and the two conditioning outputs to the
+   sampler's positive and negative inputs.
+3. Enter both texts, save a pair, then keep positive unchanged and change negative.
+   Recent should show both combinations with distinct negative excerpts. Selecting
+   one previews both; **Use pair** loads both together.
+4. Test category/tag autocomplete and metadata opt-in, resizing, light/dark,
+   switching Classic / Nodes 2.0 both ways, and workflow reload. Connected fields
+   stay authoritative; run the node to save connected metadata values.
+5. Queue the same recipe twice and change only negative once. Usage should count
+   dequeued jobs even when ComfyUI caches the node. Prepend, append and LoRA triggers
+   affect positive only. Gallery text is labelled as recorded during execution or
+   recovered from workflow metadata.
+6. Run an output metadata scan to enrich existing images. Repeat it: pairs and
+   links must remain deduplicated, with unchanged results rather than new links.
+   Conflicting or ambiguous files need review and are preserved.
+
+Fresh executions embed bounded text evidence in PNG metadata. A cached node may
+not emit fresh evidence, so its graph-derived text is explicitly marked inferred.
+Multiple distinct batch records without an image anchor remain unresolved.
+Rescanning after interruption is safe through idempotent per-file writes; this
+version does not persist a filesystem scan cursor across server restarts.
+The API can request cancellation with `POST /prompt_manager/scan/cancel`; the
+existing dashboard Hide button continues to hide progress without cancelling.
+
+Pair APIs are under `/prompt_manager/pairs`: POST saves a combination, GET lists
+and filters, GET/PATCH/DELETE `/{id}` reads, edits metadata or removes an unused
+pair. Changing text creates another combination. Null metadata updates preserve
+existing values; empty strings/lists clear supported fields. Archive a pair with
+`PATCH {"archived":true}`. GET `/export` provides versioned JSON containing both
+texts and pair metadata; this is an export format, not a JSON import endpoint.
+
+## Gallery redesign concept retained
+
+The maintainer approved the interactive
+[negative variant timeline concept](concepts/prompt-variant-timeline.html) on
+2026-10-17. Preserve it for the later gallery redesign; it is not production UI
+and does not expand the current node work.
+
+Keep one positive parent and one shared gallery, with red branches for reusable
+negative variants and smaller dots for repeat runs. Show the selected negative
+in full directly beneath the unchanged positive; selecting another variant or
+run updates that text and filters the shared gallery. Comparison shows both
+negatives. Distinguish explicitly empty negatives from an all-variants view and
+saved variants from executed runs. Production history needs persisted execution
+records; the prototype's timeline uses illustrative sample data.
+
+The Admin upgrade/scan notice remains a separate follow-up: explain automatic
+schema migration and offer an optional metadata enrichment scan or Later.

docs/screenshots/208/README.md

@@ -0,0 +1,13 @@
+# Native Library visual evidence
+
+Captured from the production workspace using disposable synthetic browser-test
+fixtures. No personal images, library, or running ComfyUI state was accessed.
+Solid-color media are intentional test images.
+
+- `library-dark.png`: grouped positive shelf, dark appearance.
+- `library-light.png`: the same shelf, light appearance.
+- `history-selected-negative.png`: selected negative directly below its positive,
+  saved-variant branches and the filtered shared gallery.
+- `history-mobile.png`: the same history at a 390-pixel viewport.
+
+These screenshots document implemented behavior, rather than a design mockup.