fix(history): honor close deadlines after early joins (#209 )

fix(history): harden context validation and worker fixtures (#209 )
test(ci): update execution context tooling fixtures (#209 )
2026-10-17 19:38:29 -07:00 · 2026-10-17 19:19:53 -07:00 · 2026-10-17 19:11:04 -07:00 · 2026-10-17 19:03:56 -07:00 · 2026-10-17 18:54:53 -07:00 · 2026-10-17 18:38:56 -07:00
276 changed files with 33710 additions and 253 deletions
@@ -0,0 +1,29 @@
+{
+  "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"
+    ]
+  },
+  "enforce_admins": true,
+  "required_pull_request_reviews": null,
+  "restrictions": null,
+  "allow_force_pushes": false,
+  "allow_deletions": false
+}
@@ -9,6 +9,9 @@ on:
  push:
    branches: [main]

+env:
+  PYTHONFAULTHANDLER: "1"
+
 jobs:
  test:
    name: py${{ matrix.python-version }} on ${{ matrix.os }}
@@ -17,7 +20,7 @@ jobs:
      fail-fast: false
      matrix:
        os: [ubuntu-latest, windows-latest, macos-latest]
-        python-version: ["3.10", "3.11", "3.12"]
+        python-version: ["3.10", "3.11", "3.12", "3.13"]

    steps:
      - uses: actions/checkout@v7
@@ -37,6 +40,83 @@ jobs:
      - name: Run tests
        run: python -m unittest discover -s tests -v

+  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
@@ -78,6 +158,19 @@ jobs:
    steps:
      - uses: actions/checkout@v7

-      # Pure-JS frontend modules, tested with the runner's preinstalled Node
+      - 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
@@ -2,22 +2,9 @@ name: Viewer.js vendor and compatibility

 on:
  pull_request:
-    paths:
-      - 'tools/vendor/**'
-      - 'tests/vendor/**'
-      - 'tests/js/**'
-      - 'web/js/**'
-      - 'web/lib/**'
-      - '.github/workflows/viewerjs-*.yml'
+    branches: [main]
  push:
    branches: [main]
-    paths:
-      - 'tools/vendor/**'
-      - 'tests/vendor/**'
-      - 'tests/js/**'
-      - 'web/js/**'
-      - 'web/lib/**'
-      - '.github/workflows/viewerjs-*.yml'
  workflow_dispatch:

 permissions:
@@ -26,8 +13,15 @@ permissions:
 jobs:
  viewerjs:
    runs-on: ubuntu-latest
+    timeout-minutes: 15
    steps:
      - uses: actions/checkout@v7
+      - uses: actions/setup-python@v7
+        with:
+          python-version: '3.12'
+          cache: pip
+          cache-dependency-path: requirements-dev.txt
+      - run: python -m pip install -r requirements-dev.txt
      - uses: actions/setup-node@v7
        with:
          node-version: '24'
@@ -36,6 +30,18 @@ jobs:
      - run: npm ci --prefix tools/vendor --ignore-scripts
      - run: npm run --prefix tools/vendor check
      - run: npm run --prefix tools/vendor test
-      - run: node --test tests/js/*.test.js
-      - run: npx --prefix tools/vendor playwright install --with-deps chromium
-      - run: npm run --prefix tools/vendor test:browser
+      - run: npx --prefix tools/vendor playwright install --with-deps chromium firefox webkit
+      - name: Frontend unit, viewer, workflow, and coverage checks
+        run: npm run --prefix tools/vendor test:coverage
+      - name: Firefox and WebKit page workflows
+        run: npm run --prefix tools/vendor test:workflows -- --project firefox --project webkit
+      - name: Upload browser traces and frontend coverage
+        if: always()
+        uses: actions/upload-artifact@v7
+        with:
+          name: frontend-test-reports
+          path: |
+            tools/vendor/test-results/
+            coverage/js/
+            !coverage/js/raw/**
+          retention-days: 14
@@ -205,3 +205,9 @@ cython_debug/
 node_modules/
 test-results/
 playwright-report/
+
+# Combined frontend coverage reports
+coverage/
+
+# Disposable upstream checkout for real-host CI
+.ci-comfyui/
@@ -142,6 +142,32 @@ _Powered by [ViewerJS](https://github.com/fengyuanchen/viewerjs) for professiona
 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
@@ -505,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

@@ -747,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:
@@ -818,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
@@ -0,0 +1,292 @@
+# Testing and merge confidence
+
+Routine changes should merge after the required checks pass and the diff is
+reviewed. Add a regression test for each bug; prove it fails before the fix.
+
+## What runs
+
+| Layer | Real components | Replaced boundaries |
+| --- | --- | --- |
+| Python tests | Domain logic, SQLite, API handlers, filesystem checks | ComfyUI host and ML runtimes where needed |
+| JavaScript tests | Frontend modules and controller methods | DOM/network fixtures for focused edge cases |
+| Viewer checks | Production Viewer.js and frontend methods in Chromium desktop/touch | Controlled images and page fixtures |
+| Real ComfyUI | Pinned CPU host, actual extension, node definitions, widgets, routes, SQLite and real execution lifecycle | Stock UI cases isolate host writes; the execution suite queues deterministic test-only nodes without models |
+| Page workflows | Production HTML, startup, click handlers, HTTP routes, SQLite, PNG files, thumbnails | ComfyUI host services and the tagging model |
+
+Page workflows cover save/edit/reload, duplicate-edit rejection, gallery image
+identity, Tags selection and viewer cleanup, delayed responses, auto-tag review
+and persistence, and infinite scrolling without replacing existing cards.
+Data safety workflows download and upload actual SQLite backups, verify restored
+prompts/tags/image links, reject corrupt uploads without changing live data, and
+exercise cancellation and network-failure recovery for restores and deletion.
+Duplicate cleanup verifies only the selected file disappears; prompt deletion
+verifies the other prompt and original image files survive.
+They run in Chromium, Firefox, WebKit, and Chromium touch emulation. No GPU,
+model downloads, running ComfyUI instance, or personal configuration is required.
+The server binds to localhost and creates disposable directories and a database;
+its reset endpoint exists only in the test harness.
+
+## Run locally
+
+Use a Python environment containing `requirements-dev.txt` and Node.js 24:
+
+```sh
+python -m unittest discover -s tests -q
+npm ci --prefix tools/vendor --ignore-scripts
+npx --prefix tools/vendor playwright install --with-deps chromium firefox webkit
+npm run --prefix tools/vendor test
+npm run --prefix tools/vendor test:coverage
+npm run --prefix tools/vendor test:workflows -- --project firefox --project webkit
+```
+
+`PM_TEST_PYTHON` selects the Python executable; `PM_TEST_PORT` selects the local
+port (default 8765). Set these environment variables when Python is outside PATH
+or the port is occupied. Test servers are never reused. On unsupported Linux
+distributions, browser system dependencies may need manual installation; CI uses
+Ubuntu. For a quick browser run, use `test:workflows -- --project chromium`.
+
+## Managed real ComfyUI
+
+The shared engine comes from an exact commit of
+[ComfyAssets Toolkit](https://github.com/ComfyAssets/ComfyAssets-Toolkit), recorded
+in `tools/vendor/package.json` and its integrity-checked lockfile. The local
+adapter sets PromptManager's database/configuration and monitoring policy;
+source discovery, coverage mapping/gates and process lifecycle are shared code.
+KikoTools remains the reference and is unchanged by this integration.
+
+Use a **dedicated disposable** ComfyUI checkout at the revision in
+`tools/vendor/comfyui-runtime.json`. With a separate Python environment, install
+that checkout's requirements, this repo's `requirements-dev.txt`, and the exact
+CPU torch/torchvision pins from the JSON (using PyTorch's CPU wheel index).
+Install Chromium using the Playwright command above, then run:
+
+```sh
+PM_COMFY_ROOT=/path/to/disposable/ComfyUI \
+PM_TEST_PYTHON=/path/to/cpu-env/bin/python \
+npm run --prefix tools/vendor test:comfyui
+```
+
+CI performs this setup automatically. `PM_COMFY_URL` can select another
+`http://127.0.0.1:PORT/` address (default port 8189). This is an owned managed
+server: a running host is never reused. Do not point it at your live checkout.
+The launcher verifies the upstream SHA and clean tracked files, snapshots only
+tracked PromptManager files, and creates temporary user/input/output/database/
+configuration/log directories. Personal untracked files are excluded. It runs
+CPU/offline with only PromptManager enabled and removes owned state on shutdown.
+The upstream checkout must also be disposable: ComfyUI may initialize local
+runtime files there. Browser host-settings writes are intercepted; actual
+PromptManager routes use the disposable host and database.
+
+`real-comfyui-reports` retains startup logs, version attachments, screenshots and
+traces for 14 days. Update upstream/frontend/CPU pins together after reviewing
+and running both startup projects; never replace the SHA with a moving branch.
+Update the toolkit via an exact reviewed commit archive and commit both npm
+manifest and lockfile. No toolkit or npm installation is needed by end users.
+
+## Coverage and failures
+
+Python CI enforces 85% combined statement/branch coverage. Frontend coverage merges
+Node V8 unit coverage with Chromium page-workflow coverage through c8. Every
+`web/js/**/*.js` and `web/js/**/*.mjs` file appears, including untouched files at zero coverage. The
+controller tests provide V8 filenames so their executed code is counted too.
+
+`tools/vendor/frontend-coverage-baseline.json` holds a floor for every file and
+metric. CI rejects regressions, missing measurements, and new files without a
+reviewed baseline. These floors reflect current coverage, with two percentage
+points of allowance for runtime/timing differences; they do not claim 80% coverage
+of the large controllers. New files need at least 80% line coverage when added via
+the baseline updater. Raise floors with `npm run --prefix tools/vendor
+coverage:update`; it reruns the suites and cannot lower an existing floor.
+Review baseline changes like code changes.
+
+The `viewerjs` check runs on every PR, including backend-only changes. GitHub
+artifacts named `frontend-test-reports` contain HTML coverage, a JSON summary,
+and failure screenshots/traces. Open traces with Playwright's `show-trace` tool.
+
+The real-host suite covers both Classic and Nodes 2.0 startup, switching both
+ways, actual database prompt selection, execution-text serialization, workflow
+reload, widget/style cleanup, node schemas and standalone-page opening. It does
+not certify every existing feature, arbitrary host versions,
+GPU/model changes, cross-pack installation or visual design changes. Those still
+need targeted validation.
+
+## Required merge checks
+
+`.github/branch-protection.json` records the required checks: all twelve Python/OS
+combinations, Python coverage, formatting, frontend modules, `viewerjs`, and `real ComfyUI compatibility`.
+The branch must be current with main, and administrators follow the same checks.
+It does not require an additional human approval.
+
+A repository administrator applies the configuration after the new workflow is
+passing, then verifies it:
+
+```sh
+gh api --method PUT repos/ComfyAssets/ComfyUI_PromptManager/branches/main/protection \
+  --input .github/branch-protection.json
+gh api repos/ComfyAssets/ComfyUI_PromptManager/branches/main/protection/required_status_checks
+```
+
+Update the manifest whenever required job names change. Keep the frontend check
+free of path filters so a required check is never silently skipped.
+
+## Unified workspace
+
+The opt-in `/prompt_manager/workspace` page opens the native Images workspace.
+Overview retains real `/prompt_manager/stats` counts. Both use the shipped toolkit. Existing admin, gallery and metadata URLs remain
+available while replacement workspaces are developed. Browser smoke tests in
+`tests/browser/workspace.spec.mjs` cover navigation/history, keyboard skip links,
+mobile layout, theme persistence/fallback, loading/retry and stale responses.
+The Images cases in `tests/browser/workspace-images.spec.mjs` additionally cover
+server-side filters, cursor append/retry, Grid/List/Masonry, bounded DOM, keyboard
+navigation, restored browsing context, Viewer.js cleanup and actual video playback.
+`tests/test_api_workspace_images.py` checks signed cursors, source identities,
+legacy compatibility and header-only dimensions. Frontend stream/layout units
+cover cancellation, malformed responses and bounded visibility lookup. These run
+automatically in the existing four-browser-project workflow and frontend coverage job. `tests/comfyui/workspace.spec.mjs` verifies the page against the
+managed host with both renderer startup configurations.
+
+Run the focused page checks with:
+
+```sh
+npm run --prefix tools/vendor test:workflows -- workspace.spec.mjs workspace-images.spec.mjs
+node --test tests/js/workspace-*.test.js
+```
+
+Use the disposable fixture and managed-host prerequisites above. A maintainer
+live check needs a ComfyUI restart after adding the Python route, then opening
+`/prompt_manager/workspace`; the mockup preview server is a separate design tool.
+
+## Prompt sending and cross-pack compatibility
+
+`tests/js/prompt-bridge-contracts.test.js` runs the Toolkit's shared protocol
+contracts against the actual checked-in browser asset. Adapter and lifecycle
+units cover role isolation, connected inputs, stale previews, callback recovery,
+serialization normalization and cached-page restoration. Browser tab workflows
+exercise real BroadcastChannel delivery; `tests/comfyui/prompt-bridge.spec.mjs`
+checks the real widgets, execution serialization and renderer transitions.
+
+The existing required real-ComfyUI job also runs `test:compat`: PromptManager
+alone, pinned KikoTools alone and both together, each starting in Classic and
+Nodes 2.0. It copies tracked source into disposable hosts, preserves independent
+configuration and checks edits/reload with both packs present. KikoTools remains
+on its current embedded foundation; this does not adopt the Toolkit in that pack.
+
+For local runs, use a clean KikoTools checkout at the revision in
+`tools/vendor/cross-pack-runtime.json`, plus the managed-host prerequisites above:
+
+```sh
+PM_COMFY_ROOT=/path/to/pinned/comfyui PM_KIKO_ROOT=/path/to/pinned/kikotools \
+  npm run --prefix tools/vendor test:compat
+```
+
+The compatibility runner reserves loopback ports 8193–8195; `PM_COMPAT_PORT`
+selects another consecutive unprivileged range. Never run automated tests against
+the maintainer's running ComfyUI. This JavaScript host extension needs a full
+browser refresh after deployment; its first slice adds no gallery send controls.
+
+### Workspace prompt inspection and sending
+
+`tests/test_api_prompt_inspector.py` covers rooted physical-file identity,
+execution/graph/parameter evidence, saved associations, ambiguity, empty negatives,
+read-only behavior, bounded PNG/video metadata and corrupt inputs. Run it with
+`python -m unittest discover -s tests -p test_api_prompt_inspector.py` in a disposable
+checkout. `tests/js/prompt-inspector.test.js` and `prompt-sender.test.js` cover
+request races, explicit destination selection, preview tickets and uncertain
+acknowledgements. Browser workflows exercise the same controls in Images and
+Viewer, including nested modal focus, mobile layouts and per-role sending.
+`tests/comfyui/prompt-sender.spec.mjs` sends through the actual workspace UI into
+real Classic and Nodes 2.0 widgets and verifies workflow serialization without
+loading a workflow or queueing generation.
+
+## Durable execution ledger
+
+`npm run --prefix tools/vendor test:executions` runs 26 real execution cases
+(13 each in Classic and Nodes 2.0, retaining the original ten ledger cases),
+alongside the existing 78 stock host cases and six pack compatibility cases. Run
+these suites serially because they share report cleanup and owned host resources. Use the same pinned disposable `PM_COMFY_ROOT`,
+`PM_TEST_PYTHON` and Playwright browser installation as `test:comfyui`:
+
+```sh
+PM_COMFY_ROOT=/path/to/disposable/ComfyUI \
+PM_TEST_PYTHON=/path/to/cpu-env/bin/python \
+PLAYWRIGHT_BROWSERS_PATH=/path/to/playwright-browsers \
+npm run --prefix tools/vendor test:executions
+```
+
+This suite reserves `http://127.0.0.1:8196/` and never attaches to an existing
+server. Each renderer worker launches a CPU/offline host with independent
+configuration, database, user and output directories. The shared launcher installs
+tracked snapshots of PromptManager and a tiny temporary Git-tracked fixture pack.
+Only that owned test pack provides cached output, deliberate failure, an
+interruptible queue gate, deterministic CLIP/model/latent inputs and control routes.
+It replaces external sampler inference and trigger injection only inside the
+disposable host; real registered PromptManager classes, queue, executor and cache
+remain active. Its cache control resets the real executor on its worker to exercise
+Classic, LRU, RAM-pressure and disabled caching; production has no test nodes or
+endpoints. Cleanup always releases the gate, stops the host and removes both
+owned directories. Logs and Playwright failures appear under
+`tools/vendor/test-results/executions/` and the existing CI artifact upload.
+
+The request-only case creates no browser or WebSocket client. Other cases verify
+real `execution_cached` node lists after a failed job, both renderer switches and
+workflow reload, repeated public UUIDs, interruption, graph rejection, and queued
+removal before dequeue. Actual host history is checked alongside finite polling
+of SQLite after the background writer. The read helper uses SQLite read-only
+mode, exact UUID job/session filters and a 256-row limit; absent ledger schema is
+explicit, while missing/corrupt databases and other read errors fail the suite.
+Sensitive diagnostic and extra-data sentinels must never appear in lifecycle
+rows. Exact prompt fields are deliberately persisted only in context snapshots.
+The same read-only helper calls the internal `get_execution_contexts` reader for
+requested run IDs and inspects their snapshot identities and current usage counts.
+
+For a meaningful historical failure, create a separate disposable detached
+checkout at `fdba6fd0126a8a40cbd6f3e50ed4c5de8991efd4` and pass its path to
+`launchExecutionHost({ repository })`. That base also predates the #211 config
+import fix: after confirming an actual headless job succeeded, use
+`readExecutionRuns` on its explicitly constructed owned path
+`host.stateDirectory/custom_nodes/promptmanager/prompts.db`. It reports
+`schemaAvailable: false` and empty rows, failing the expected succeeded-row
+assertion. This historical inspection stays outside the ordinary suite, which
+requires the configured database path. The launcher also accepts
+`PM_EXECUTION_REPOSITORY`; omit it for the implementation worktree. Stage new
+files first because snapshots use `git ls-files`. Never reset the implementation
+checkout or point this harness at a live ComfyUI checkout.
+
+The package-qualified config import also corrects #211: the configured database
+path is honored when PromptManager runs as a ComfyUI custom node. Follow the
+[database path compatibility and safe backup sequence](README.md#database-path-compatibility-211)
+before restarting an existing deployment.
+
+The ledger records only new dequeued jobs and observed outcomes. Existing
+counters, prompts, pairs and image links are preserved, with no reconstructed old
+runs. Schema v2 migrates automatically when the configured database opens, adding
+immutable prompt-context tables without backfilling old runs. The internal reader
+returns `not_captured` for preexisting runs; a new capture can instead be complete,
+partial, ambiguous or unavailable. Keep the existing database backup/restore
+procedure: restored history changes storage epoch, and old library navigation
+hints must never be rebound to unrelated current rows.
+
+Context cases verify A,A,B as three jobs/two combinations/one positive snapshot
+when every positive field matches; all five registered PM variants; exact raw
+whitespace and Unicode; known empty negatives; Text negative roles and crossed
+pairs; fully cached repeats after trigger configuration changes; successful source
+text before downstream failure/interruption; unknown cached dynamic raw; unsupported
+transforms and duplicate list invocation ambiguity. Fresh values use
+`runtime_return`, cached effective STRING outputs use `host_cache_output`, and
+submitted literals use graph inference. SQL NULL means unknown; empty text is
+known. Text is never trimmed or normalized by capture. Cache availability and
+inferred sampler roles prove neither sampler execution nor consumption, and list
+indices do not identify generated images. External trigger configuration changes
+must not rewrite the effective text of an existing cached result. The cache matrix
+also checks that positive/pair usage increments once per dequeued job while a
+negative-only component retains its existing zero standalone-use count.
+
+Fault injection for restore fencing, inherited async capability revocation, queued
+finalization, epoch and generation revocation, provider-local cache entries, eviction/reset, malformed or
+oversized payloads and missing host interfaces lives in the focused Python
+`test_execution*.py` suites. These use disposable storage and mocked host interfaces;
+they are distinct from the pinned real executor/cache checks above. No output-file
+associations or chronological timeline are introduced. Deploying the Python changes
+requires a ComfyUI restart. After restart, queue three new jobs A,A,B with identical
+positive fields and two negative recipes; existing counters should advance as usual,
+and internal context inspection should find three attempts and two combinations.
@@ -3,12 +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
+- 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.

@@ -38,18 +40,29 @@ def get_version():
    return "unknown"


-from .prompt_manager import PromptManager  # noqa: E402
+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",
 }
@@ -145,6 +158,34 @@ except Exception as e:
            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
@@ -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
@@ -0,0 +1,376 @@
+"""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 _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()
@@ -0,0 +1,194 @@
+"""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")
+_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("PRAGMA busy_timeout=250")
+        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
@@ -0,0 +1,313 @@
+"""Versioned additive schema for observed execution attempts."""
+
+import re
+import sqlite3
+from uuid import uuid4
+
+SCHEMA_VERSION = 2
+
+
+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 = 2",
+    _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
+
+    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):
+        return "Unsupported execution history metadata/version"
+    version = metadata[0][0]
+    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" and version == 1:
+            checks = tuple(
+                check.replace("schema_version = 2", "schema_version = 1")
+                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:
+        return context_schema_error(conn)
+    return None
+
+
+def install_context_schema(conn: sqlite3.Connection) -> None:
+    from .execution_context_schema import install_context_schema as install
+
+    install(conn)
+
+
+def migrate_execution_history(conn: sqlite3.Connection) -> None:
+    """Install versioned history and context tables 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)
+        else:
+            version, epoch = conn.execute(
+                "SELECT schema_version,storage_epoch FROM execution_history_metadata"
+            ).fetchone()
+            if version == 1:
+                conn.execute(
+                    "ALTER TABLE execution_history_metadata RENAME TO "
+                    "execution_history_metadata_v1"
+                )
+                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,2,?)", (epoch,)
+                )
+                conn.execute("DROP TABLE execution_history_metadata_v1")
+                install_context_schema(conn)
+        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
@@ -0,0 +1,321 @@
+"""Read-only positive-parent views over existing prompt and pair tables."""
+
+_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))"""
+_ORDERS = {
+    "newest": ("COALESCE(julianday(p.created_at),0)", "DESC"),
+    "oldest": ("COALESCE(julianday(p.created_at),0)", "ASC"),
+    "name": ("LOWER(p.text)", "ASC"),
+    "rating": ("COALESCE(p.rating,0)", "DESC"),
+}
+
+
+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 _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 '\\'")
+        parent_params.append(f"%{_escape(text)}%")
+        pair.append(
+            "(p.text LIKE ? ESCAPE '\\' OR n.text LIKE ? ESCAPE '\\' "
+            "OR pp.title LIKE ? ESCAPE '\\')"
+        )
+        pair_params.extend([f"%{_escape(text)}%"] * 3)
+    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."""
+
+    def _library_parents(
+        self,
+        where: str,
+        params: list,
+        suffix: str = "",
+        sort_expression: str | None = None,
+    ) -> list[dict]:
+        conn = self.model.get_connection()
+        rows = conn.execute(
+            "SELECT p.id,p.text,p.category,p.rating,p.created_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 prompts p 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"])
+        return result
+
+    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:
+        _limit(limit)
+        if sort not in _ORDERS:
+            raise ValueError("Unknown library sort order")
+        where, params = _filter(text, category, [] if tags is None else tags)
+        conn = self.model.get_connection()
+        ceiling = conn.execute("SELECT COALESCE(MAX(id),0) FROM prompts").fetchone()[0]
+        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)
+        total = conn.execute(
+            "SELECT COUNT(*) FROM prompts p WHERE " + where, params
+        ).fetchone()[0]
+        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(
+            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 {
+            "items": rows,
+            "total": total,
+            "has_more": has_more,
+            "next_key": next_key,
+        }
+
+    def get_library_positive(
+        self,
+        positive_id: int,
+        *,
+        variant_limit: int = 40,
+        variant_after: int | None = None,
+    ) -> dict | None:
+        _limit(variant_limit)
+        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(_ELIGIBLE + " AND p.id=?", [positive_id])
+        if not rows:
+            return None
+        conn = self.model.get_connection()
+        where = "pp.positive_prompt_id=? AND pp.archived=0"
+        params = [positive_id]
+        variant_total = conn.execute(
+            "SELECT COUNT(*) FROM prompt_pairs pp WHERE " + where, params
+        ).fetchone()[0]
+        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.negative_prompt_id IS NULL AS explicit_empty,"
+                "(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["explicit_empty"] = bool(row["explicit_empty"])
+        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 {
+            "positive": rows[0],
+            "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_options(self) -> dict:
+        rows = (
+            self.model.get_connection()
+            .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 {"categories": [row["category"] for row in rows]}
+
+    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,
+        positive_id: int,
+        *,
+        pair_id: int | None = None,
+        limit: int = 40,
+        after: int | None = None,
+    ) -> dict:
+        _limit(limit)
+        conn = self.model.get_connection()
+        if (
+            pair_id is not None
+            and not conn.execute(
+                "SELECT 1 FROM prompt_pairs WHERE id=? "
+                "AND positive_prompt_id=? AND archived=0",
+                (pair_id, positive_id),
+            ).fetchone()
+        ):
+            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:
+            if type(after) is not int or after <= 0:
+                raise ValueError("Invalid image cursor")
+            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 {
+            "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,
+        }
@@ -4,14 +4,23 @@ Database schema and models for KikoTextEncode prompt storage.

 import contextlib
 import datetime
+import json
+import math
 import shutil
 import sqlite3
 import os
 import threading
+import tempfile
 import weakref
 from contextlib import closing
 from pathlib import Path
-from typing import Set, Tuple
+from typing import Optional, Set, Tuple
+
+from .execution_history_schema import (
+    history_schema_error,
+    migrate_execution_history,
+    rotate_history_epoch,
+)

 # Import logging system
 try:
@@ -29,6 +38,41 @@ except ImportError:
 ISO_TIMESTAMP_FORMAT = "%Y-%m-%dT%H:%M:%f+00:00"


+MAX_PAIR_PROVENANCE_LENGTH = 4_000_000
+MAX_PAIR_PROVENANCE_DEPTH = 64
+
+
+def pair_provenance_error(encoded: object) -> Optional[str]:
+    """Validate the same bounded JSON object for live writes and restores."""
+    if not isinstance(encoded, str):
+        return "Pair provenance must be JSON text"
+    if len(encoded) > MAX_PAIR_PROVENANCE_LENGTH:
+        return "Pair provenance exceeds the size limit"
+    try:
+        value = json.loads(encoded)
+    except (ValueError, TypeError, RecursionError):
+        return "Pair provenance is not readable JSON"
+    if not isinstance(value, dict):
+        return "Pair provenance must be a JSON object"
+    pending = [(iter((value,)), 0)]
+    while pending:
+        children, depth = pending[-1]
+        try:
+            item = next(children)
+        except StopIteration:
+            pending.pop()
+            continue
+        if depth > MAX_PAIR_PROVENANCE_DEPTH:
+            return "Pair provenance exceeds the nesting limit"
+        if isinstance(item, float) and not math.isfinite(item):
+            return "Pair provenance contains nonfinite numbers"
+        if isinstance(item, dict):
+            pending.append((iter(item.values()), depth + 1))
+        elif isinstance(item, list):
+            pending.append((iter(item), depth + 1))
+    return None
+
+
 def utc_now_iso() -> str:
    """Current UTC time as ISO text with microseconds.

@@ -235,6 +279,225 @@ class PromptModel:

        # Image uniqueness by full path instead of basename (3.2.4)
        self._migrate_image_uniqueness_by_path(conn)
+        self._migrate_prompt_pairs(conn)
+        migrate_execution_history(conn)
+
+    def _migrate_prompt_pairs(self, conn: sqlite3.Connection) -> None:
+        conn.execute("SAVEPOINT prompt_pair_schema")
+        try:
+            conn.execute("""
+                CREATE TABLE IF NOT EXISTS prompt_pairs (
+                    id INTEGER PRIMARY KEY AUTOINCREMENT,
+                    positive_prompt_id INTEGER NOT NULL,
+                    negative_prompt_id INTEGER,
+                    title TEXT,
+                    category TEXT,
+                    rating INTEGER CHECK(rating BETWEEN 1 AND 5),
+                    notes TEXT,
+                    created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
+                    updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
+                    last_used_at TEXT,
+                    run_count INTEGER NOT NULL DEFAULT 0,
+                    archived INTEGER NOT NULL DEFAULT 0 CHECK(archived IN (0, 1)),
+                    FOREIGN KEY(positive_prompt_id)
+                        REFERENCES prompts(id) ON DELETE RESTRICT,
+                    FOREIGN KEY(negative_prompt_id)
+                        REFERENCES prompts(id) ON DELETE RESTRICT,
+                    UNIQUE(positive_prompt_id, negative_prompt_id)
+                )
+            """)
+            conn.execute("""
+                CREATE TABLE IF NOT EXISTS prompt_pair_tags (
+                    pair_id INTEGER NOT NULL,
+                    tag_id INTEGER NOT NULL,
+                    PRIMARY KEY(pair_id, tag_id),
+                    FOREIGN KEY(pair_id) REFERENCES prompt_pairs(id) ON DELETE CASCADE,
+                    FOREIGN KEY(tag_id) REFERENCES tags(id) ON DELETE CASCADE
+                )
+            """)
+            conn.execute("""
+                CREATE TABLE IF NOT EXISTS image_prompt_pairs (
+                    id INTEGER PRIMARY KEY AUTOINCREMENT,
+                    image_id INTEGER NOT NULL,
+                    pair_id INTEGER NOT NULL,
+                    provenance TEXT NOT NULL,
+                    created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
+                    FOREIGN KEY(image_id)
+                        REFERENCES generated_images(id) ON DELETE CASCADE,
+                    FOREIGN KEY(pair_id) REFERENCES prompt_pairs(id) ON DELETE RESTRICT,
+                    UNIQUE(pair_id, image_id, provenance)
+                )
+            """)
+            self._create_pair_indexes(conn)
+            error = self._pair_schema_error(conn)
+            if error:
+                raise sqlite3.DatabaseError(error)
+            conn.execute("RELEASE prompt_pair_schema")
+        except BaseException:
+            conn.execute("ROLLBACK TO prompt_pair_schema")
+            conn.execute("RELEASE prompt_pair_schema")
+            raise
+
+    @staticmethod
+    def _create_pair_indexes(conn: sqlite3.Connection) -> None:
+        conn.execute(
+            "CREATE UNIQUE INDEX IF NOT EXISTS idx_prompt_pairs_positive_only "
+            "ON prompt_pairs(positive_prompt_id) WHERE negative_prompt_id IS NULL"
+        )
+        for name, target in (
+            ("idx_prompt_pairs_negative", "prompt_pairs(negative_prompt_id)"),
+            ("idx_prompt_pair_tags_tag", "prompt_pair_tags(tag_id)"),
+            ("idx_image_prompt_pairs_image", "image_prompt_pairs(image_id)"),
+        ):
+            conn.execute(f"CREATE INDEX IF NOT EXISTS {name} ON {target}")
+
+    @staticmethod
+    def _pair_schema_error(conn: sqlite3.Connection) -> Optional[str]:
+        required = {
+            "prompt_pairs": {
+                "id",
+                "positive_prompt_id",
+                "negative_prompt_id",
+                "title",
+                "category",
+                "rating",
+                "notes",
+                "created_at",
+                "updated_at",
+                "last_used_at",
+                "run_count",
+                "archived",
+            },
+            "prompt_pair_tags": {"pair_id", "tag_id"},
+            "image_prompt_pairs": {
+                "id",
+                "image_id",
+                "pair_id",
+                "provenance",
+                "created_at",
+            },
+        }
+        present = {
+            row[0]
+            for row in conn.execute("SELECT name FROM sqlite_master WHERE type='table'")
+        }
+        if not present.intersection(required):
+            return None
+        if not set(required).issubset(present):
+            return "Incomplete prompt pair schema"
+        foreign_keys = {
+            "prompt_pairs": {
+                ("positive_prompt_id", "prompts", "id", "RESTRICT"),
+                ("negative_prompt_id", "prompts", "id", "RESTRICT"),
+            },
+            "prompt_pair_tags": {
+                ("pair_id", "prompt_pairs", "id", "CASCADE"),
+                ("tag_id", "tags", "id", "CASCADE"),
+            },
+            "image_prompt_pairs": {
+                ("pair_id", "prompt_pairs", "id", "RESTRICT"),
+                ("image_id", "generated_images", "id", "CASCADE"),
+            },
+        }
+        unique = {
+            "prompt_pairs": ("positive_prompt_id", "negative_prompt_id"),
+            "prompt_pair_tags": ("pair_id", "tag_id"),
+            "image_prompt_pairs": ("pair_id", "image_id", "provenance"),
+        }
+        non_null = {
+            "prompt_pairs": {
+                "positive_prompt_id",
+                "created_at",
+                "updated_at",
+                "run_count",
+                "archived",
+            },
+            "prompt_pair_tags": {"pair_id", "tag_id"},
+            "image_prompt_pairs": {"pair_id", "image_id", "provenance", "created_at"},
+        }
+        integer_columns = {
+            "id",
+            "positive_prompt_id",
+            "negative_prompt_id",
+            "pair_id",
+            "tag_id",
+            "image_id",
+            "rating",
+            "run_count",
+            "archived",
+        }
+        for table, columns in required.items():
+            info = {row[1]: row for row in conn.execute(f"PRAGMA table_info({table})")}
+            if not columns.issubset(info):
+                return f"{table} is missing required pair columns"
+            if any(not info[name][3] for name in non_null[table]):
+                return f"{table} has nullable required pair columns"
+            if any(
+                info[name][2].upper()
+                != ("INTEGER" if name in integer_columns else "TEXT")
+                for name in columns
+            ):
+                return f"{table} has invalid pair column types"
+            primary = tuple(
+                name
+                for name in sorted(info, key=lambda name: info[name][5])
+                if info[name][5]
+            )
+            if primary != (
+                ("pair_id", "tag_id") if table == "prompt_pair_tags" else ("id",)
+            ):
+                return f"{table} has an invalid pair primary key"
+            actual_fks = {
+                (row[3], row[2], row[4], row[6])
+                for row in conn.execute(f"PRAGMA foreign_key_list({table})")
+            }
+            if actual_fks != foreign_keys[table]:
+                return f"{table} has invalid pair foreign key definitions"
+            if conn.execute(f"PRAGMA foreign_key_check({table})").fetchone():
+                return f"{table} contains foreign key violations"
+            indexes = list(conn.execute(f"PRAGMA index_list({table})"))
+            index_columns = {}
+            for row in indexes:
+                name = row[1].replace('"', '""')
+                index_columns[row[1]] = tuple(
+                    item[2] for item in conn.execute(f'PRAGMA index_info("{name}")')
+                )
+            if not any(
+                row[2] and not row[4] and index_columns[row[1]] == unique[table]
+                for row in indexes
+            ):
+                return f"{table} is missing pair uniqueness constraints"
+            if table == "prompt_pairs":
+                null_unique = False
+                for row in indexes:
+                    if (
+                        row[2]
+                        and row[4]
+                        and index_columns[row[1]] == ("positive_prompt_id",)
+                    ):
+                        sql = conn.execute(
+                            "SELECT sql FROM sqlite_master WHERE type='index' AND name"
+                            "=?",
+                            (row[1],),
+                        ).fetchone()[0]
+                        if " ".join(sql.lower().split()).endswith(
+                            "where negative_prompt_id is null"
+                        ):
+                            null_unique = True
+                if not null_unique:
+                    return "prompt_pairs is missing empty-negative pair uniqueness"
+        if conn.execute(
+            "SELECT 1 FROM image_prompt_pairs ip "
+            "JOIN prompt_pairs pp ON pp.id=ip.pair_id "
+            "JOIN generated_images gi ON gi.id=ip.image_id "
+            "WHERE pp.positive_prompt_id IS NOT gi.prompt_id LIMIT 1"
+        ).fetchone():
+            return "Image pair association has a different positive prompt owner"
+        for row in conn.execute("SELECT provenance FROM image_prompt_pairs"):
+            error = pair_provenance_error(row[0])
+            if error:
+                return error
+        return None

    def _create_indexes(self, conn: sqlite3.Connection) -> None:
        """
@@ -772,6 +1035,18 @@ class PromptModel:
                    False,
                    "prompts table is missing required columns: " + ", ".join(missing),
                )
+            forbidden = conn.execute(
+                "SELECT type FROM sqlite_master WHERE type IN ('trigger', 'view') "
+                "LIMIT 1"
+            ).fetchone()
+            if forbidden:
+                return False, f"Database contains an unsupported {forbidden[0]}"
+            pair_error = PromptModel._pair_schema_error(conn)
+            if pair_error:
+                return False, pair_error
+            history_error = history_schema_error(conn)
+            if history_error:
+                return False, history_error
            return True, "ok"
        except sqlite3.Error as e:
            return False, f"Invalid SQLite database: {e}"
@@ -814,13 +1089,11 @@ class PromptModel:
        """
        Replace the live database with the file at src_path.

-        The current database is first backed up next to itself as
-        ``<db_path>.backup_<YYYYmmdd_HHMMSS>`` (WAL content included), the
-        pages of ``src_path`` are copied into the live database through
-        SQLite's backup API (visible to every open connection) and the schema
-        migrations run on the restored content.
-
-        Callers should run verify_database_file(src_path) first.
+        A disposable SQLite backup of the source is migrated and verified first.
+        Only then is the current database backed up next to itself as
+        ``<db_path>.backup_<YYYYmmdd_HHMMSS>`` (WAL content included), and the
+        staged pages replace the live pages through SQLite's backup API.
+        Migration failures never replace live content or modify the source.

        Args:
            src_path: SQLite file to restore
@@ -833,17 +1106,46 @@ class PromptModel:
            RuntimeError: If the previous database could not be backed up
            OSError: If the restored file could not be copied into place
        """
-        backup_path = ""
-        if os.path.exists(self.db_path):
-            timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
-            backup_path = f"{self.db_path}.backup_{timestamp}"
-            if not self.backup_database(backup_path):
-                raise RuntimeError("Could not back up the current database")
-            self.logger.info(f"Current database backed up to: {backup_path}")
+        ok, reason = self.verify_database_file(src_path)
+        if not ok:
+            raise ValueError(reason)
+        with tempfile.TemporaryDirectory(prefix="prompt-manager-restore-") as directory:
+            staged_path = str(Path(directory) / "staged.db")
+            source = sqlite3.connect(
+                Path(src_path).resolve().as_uri() + "?mode=ro", uri=True
+            )
+            try:
+                with closing(sqlite3.connect(staged_path)) as destination:
+                    source.backup(destination)
+            finally:
+                source.close()
+            staged = None
+            try:
+                staged = PromptModel(staged_path)
+                rotate_history_epoch(staged.get_connection())
+                ok, reason = self.verify_database_file(staged_path)
+                if not ok:
+                    raise ValueError(reason)
+                staged.close_all()
+                backup_path = ""
+                if os.path.exists(self.db_path):
+                    timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
+                    backup_path = f"{self.db_path}.backup_{timestamp}"
+                    if not self.backup_database(backup_path):
+                        raise RuntimeError("Could not back up the current database")
+                    self.logger.info(f"Current database backed up to: {backup_path}")
+                self._replace_contents_from(staged_path)
+                with PromptModel._init_lock:
+                    PromptModel._initialized_paths.add(self._schema_key())
+                self.logger.info(f"Database restored from {src_path}")
+                from .execution_history import notify_history_restore

-        self._replace_contents_from(src_path)
-        with PromptModel._init_lock:
-            PromptModel._initialized_paths.discard(self._schema_key())
-        self._ensure_database_exists()
-        self.logger.info(f"Database restored from {src_path}")
-        return backup_path
+                notify_history_restore(self.db_path)
+                return backup_path
+            finally:
+                if staged is not None:
+                    staged.close_all()
+                with PromptModel._init_lock:
+                    PromptModel._initialized_paths.discard(
+                        os.path.normcase(os.path.realpath(staged_path))
+                    )
@@ -9,6 +9,10 @@ import os
 from typing import Optional, List, Dict, Any

 from .models import PromptModel, normalize_image_path, utc_now_iso
+from .pairs import PairDependencyError, PairOperations
+from .library import LibraryOperations
+from .execution_history import ExecutionHistoryOperations
+from .execution_contexts import ExecutionContextOperations

 # Import logging system
 try:
@@ -84,7 +88,10 @@ def _resolve_db_path(db_path: Optional[str] = None) -> str:

    if db_path is None:
        try:
-            from py.config import PromptManagerConfig
+            try:
+                from ..py.config import PromptManagerConfig
+            except ImportError:
+                from py.config import PromptManagerConfig

            db_path = PromptManagerConfig.DEFAULT_DB_PATH
        except Exception:
@@ -146,7 +153,12 @@ def _coerce_file_info(metadata: Any) -> tuple:
    )


-class PromptDatabase:
+class PromptDatabase(
+    PairOperations,
+    LibraryOperations,
+    ExecutionHistoryOperations,
+    ExecutionContextOperations,
+):
    """Database operations class for managing prompts."""

    def __init__(self, db_path: Optional[str] = None):
@@ -581,7 +593,8 @@ class PromptDatabase:
        Returns:
            bool: True if deletion was successful, False otherwise
        """
-        with self.model.get_connection() as conn:
+        with self._pair_transaction() as conn:
+            self._ensure_prompts_unpaired(conn, [prompt_id])
            # First delete related images to avoid foreign key constraint
            conn.execute(
                "DELETE FROM generated_images WHERE prompt_id = ?", (prompt_id,)
@@ -597,7 +610,7 @@ class PromptDatabase:
        Returns:
            Number of prompts deleted.
        """
-        with self.model.get_connection() as conn:
+        with self._pair_transaction() as conn:
            # Get IDs first for cascade cleanup
            ids = [
                r[0]
@@ -607,6 +620,7 @@ class PromptDatabase:
            ]
            if not ids:
                return 0
+            self._ensure_prompts_unpaired(conn, ids)
            placeholders = ",".join("?" * len(ids))
            conn.execute(
                f"DELETE FROM generated_images WHERE prompt_id IN ({placeholders})",
@@ -632,7 +646,9 @@ class PromptDatabase:
        """
        with self.model.get_connection() as conn:
            cursor = conn.execute(
-                "SELECT DISTINCT TRIM(category) as category FROM prompts WHERE "
+                "SELECT TRIM(category) AS category FROM prompts WHERE "
+                "category IS NOT NULL AND TRIM(category) != '' UNION "
+                "SELECT TRIM(category) AS category FROM prompt_pairs WHERE "
                "category IS NOT NULL AND TRIM(category) != '' ORDER BY category"
            )
            return [row["category"] for row in cursor.fetchall()]
@@ -699,16 +715,16 @@ class PromptDatabase:

    def get_all_tags(self) -> List[str]:
        """
-        Get all unique tags that are in use (linked to at least one prompt).
+        Get all unique tags linked to a standalone prompt or reusable pair.

        Returns:
            Sorted list of tag names
        """
        with self.model.get_connection() as conn:
            cursor = conn.execute(
-                "SELECT DISTINCT t.name AS tag "
-                "FROM prompt_tags pt "
-                "JOIN tags t ON pt.tag_id = t.id "
+                "SELECT t.name AS tag FROM tags t WHERE "
+                "EXISTS(SELECT 1 FROM prompt_tags pt WHERE pt.tag_id=t.id) OR "
+                "EXISTS(SELECT 1 FROM prompt_pair_tags pp WHERE pp.tag_id=t.id) "
                "ORDER BY t.name"
            )
            return [row["tag"] for row in cursor.fetchall()]
@@ -784,7 +800,8 @@ class PromptDatabase:

        Returns tags whose name starts with ``prefix`` (case-insensitive),
        counted over the prompts that carry every tag in ``with_tags``; the
-        context tags themselves are left out. Ordered by count, then name.
+        context tags themselves are left out. Counts include separate standalone
+        and pair references, with owner types kept distinct. Ordered by count/name.
        """
        seen = set()
        context = []
@@ -799,11 +816,11 @@ class PromptDatabase:
        if context:
            placeholders = ",".join("?" * len(context))
            where.append(
-                "pt.prompt_id IN ("
-                " SELECT pt2.prompt_id FROM prompt_tags pt2"
+                "(pt.kind,pt.owner_id) IN ("
+                " SELECT pt2.kind,pt2.owner_id FROM tag_refs pt2"
                " JOIN tags t2 ON t2.id = pt2.tag_id"
                f" WHERE t2.name COLLATE NOCASE IN ({placeholders})"
-                " GROUP BY pt2.prompt_id"
+                " GROUP BY pt2.kind,pt2.owner_id"
                " HAVING COUNT(DISTINCT lower(t2.name)) = ?)"
            )
            params.extend(context)
@@ -813,8 +830,11 @@ class PromptDatabase:
        params.append(max(1, int(limit)))

        sql = (
+            "WITH tag_refs(kind,owner_id,tag_id) AS ("
+            "SELECT 'prompt',prompt_id,tag_id FROM prompt_tags UNION ALL "
+            "SELECT 'pair',pair_id,tag_id FROM prompt_pair_tags) "
            "SELECT t.name AS name, COUNT(*) AS count"
-            " FROM prompt_tags pt JOIN tags t ON t.id = pt.tag_id"
+            " FROM tag_refs pt JOIN tags t ON t.id = pt.tag_id"
            f" WHERE {' AND '.join(where)}"
            " GROUP BY t.id"
            " ORDER BY count DESC, t.name COLLATE NOCASE ASC"
@@ -921,6 +941,8 @@ class PromptDatabase:

        old_name = old_name.strip()
        new_name = new_name.strip()
+        if old_name == new_name:
+            return {"success": True, "affected_count": 0, "skipped_count": 0}

        with self.model.get_connection() as conn:
            cursor = conn.execute("SELECT id FROM tags WHERE name = ?", (old_name,))
@@ -942,12 +964,14 @@ class PromptDatabase:
            if existing_new:
                # Target tag exists — merge: move links, handle conflicts, delete old
                new_tag_id = existing_new["id"]
-                conn.execute(
-                    "UPDATE OR IGNORE prompt_tags SET tag_id = ? WHERE tag_id = ?",
-                    (new_tag_id, old_tag_id),
-                )
-                # Delete remaining links (conflicts = prompt already had the target tag)
-                conn.execute("DELETE FROM prompt_tags WHERE tag_id = ?", (old_tag_id,))
+                for junction in ("prompt_tags", "prompt_pair_tags"):
+                    conn.execute(
+                        f"UPDATE OR IGNORE {junction} SET tag_id = ? WHERE tag_id = ?",
+                        (new_tag_id, old_tag_id),
+                    )
+                    conn.execute(
+                        f"DELETE FROM {junction} WHERE tag_id = ?", (old_tag_id,)
+                    )
                conn.execute("DELETE FROM tags WHERE id = ?", (old_tag_id,))
            else:
                # Simple rename
@@ -966,7 +990,7 @@ class PromptDatabase:
        """
        Remove a tag from all prompts that use it.

-        With junction tables, deleting from the tags table cascades to prompt_tags.
+        Deleting from tags cascades to standalone and pair tag relationships.

        Args:
            tag_name: Tag name to remove
@@ -1033,20 +1057,27 @@ class PromptDatabase:
                if not src_row:
                    continue
                src_id = src_row["id"]
+                if src_id == target_id:
+                    continue

                cursor = conn.execute(
                    "SELECT COUNT(*) as c FROM prompt_tags WHERE tag_id = ?", (src_id,)
                )
                src_count = cursor.fetchone()["c"]

-                if src_count > 0:
-                    # Move links from source to target (ignore conflicts)
-                    conn.execute(
-                        "UPDATE OR IGNORE prompt_tags SET tag_id = ? WHERE tag_id = ?",
-                        (target_id, src_id),
-                    )
-                    # Delete remaining (conflicts = prompt already had target tag)
-                    conn.execute("DELETE FROM prompt_tags WHERE tag_id = ?", (src_id,))
+                pair_count = conn.execute(
+                    "SELECT COUNT(*) FROM prompt_pair_tags WHERE tag_id = ?", (src_id,)
+                ).fetchone()[0]
+                if src_count or pair_count:
+                    for junction in ("prompt_tags", "prompt_pair_tags"):
+                        conn.execute(
+                            f"UPDATE OR IGNORE {junction} "
+                            "SET tag_id = ? WHERE tag_id = ?",
+                            (target_id, src_id),
+                        )
+                        conn.execute(
+                            f"DELETE FROM {junction} WHERE tag_id = ?", (src_id,)
+                        )
                    tags_merged += 1
                    affected += src_count

@@ -1290,7 +1321,7 @@ class PromptDatabase:
        """
        self.logger.info("Starting duplicate cleanup process with image preservation")
        try:
-            with self.model.get_connection() as conn:
+            with self._pair_transaction() as conn:
                # Find duplicates by text content (case-insensitive)
                # Note: Removed ORDER BY from GROUP_CONCAT for SQLite compatibility
                # We'll sort the IDs manually after fetching
@@ -1319,6 +1350,11 @@ class PromptDatabase:
                    id_date_pairs.sort(key=lambda x: x[1])  # Sort by date
                    sorted_ids = [int(pair[0]) for pair in id_date_pairs]

+                    try:
+                        self._ensure_prompts_unpaired(conn, sorted_ids)
+                    except PairDependencyError:
+                        continue
+
                    # Keep the oldest one (first), merge and delete the rest
                    primary_id = sorted_ids[0]  # Keep the oldest
                    duplicate_ids = sorted_ids[1:]
@@ -1864,7 +1900,8 @@ class PromptDatabase:
        Raises:
            sqlite3.IntegrityError: Another prompt already has the new hash.
        """
-        with self.model.get_connection() as conn:
+        with self._pair_transaction() as conn:
+            self._ensure_prompts_unpaired(conn, [prompt_id])
            cursor = conn.execute(
                "UPDATE prompts SET text = ?, hash = ?, updated_at = ? WHERE id = ?",
                (
@@ -1912,7 +1949,8 @@ class PromptDatabase:
            Number of prompts actually deleted.
        """
        count = 0
-        with self.model.get_connection() as conn:
+        with self._pair_transaction() as conn:
+            self._ensure_prompts_unpaired(conn, prompt_ids)
            for pid in prompt_ids:
                conn.execute("DELETE FROM generated_images WHERE prompt_id = ?", (pid,))
                cursor = conn.execute("DELETE FROM prompts WHERE id = ?", (pid,))
@@ -1985,7 +2023,7 @@ class PromptDatabase:
            cursor = conn.execute(
                """SELECT gi.prompt_id, p.text, p.category, p.rating, p.notes,
                          gi.workflow_data, gi.prompt_metadata, gi.generation_time,
-                          gi.parameters
+                          gi.parameters, gi.id, gi.image_path, gi.file_path
                   FROM generated_images gi
                   JOIN prompts p ON gi.prompt_id = p.id
                   WHERE gi.image_path = ? OR gi.file_path = ?
@@ -1997,9 +2035,71 @@ class PromptDatabase:
                    f"%{escape_like(os.path.basename(image_path))}",
                ),
            )
-            row = cursor.fetchone()
-            if not row:
+            rows = cursor.fetchall()
+            exact = [
+                row
+                for row in rows
+                if row[10] == image_path or row[11] == normalize_image_path(image_path)
+            ]
+            matches = exact or rows
+            if len(matches) != 1:
                return None
+            row = matches[0]
+            associations = self.get_image_prompt_pairs(row[9])
+            pair_ids = {item["pair_id"] for item in associations}
+            pair = (
+                self.get_prompt_pair(next(iter(pair_ids)))
+                if len(pair_ids) == 1
+                else None
+            )
+            candidates = []
+            for item in associations:
+                evidence = item["provenance"]
+                positive_side = evidence.get("positive")
+                negative_side = evidence.get("negative")
+                if not isinstance(positive_side, dict) or not isinstance(
+                    negative_side, dict
+                ):
+                    continue
+                positive, negative = positive_side.get("text"), negative_side.get(
+                    "text"
+                )
+                if not isinstance(positive, str) or not isinstance(negative, str):
+                    continue
+                sources = []
+                for side in (positive_side, negative_side):
+                    provenance = side.get("provenance")
+                    sources.append(
+                        provenance.get("text_source")
+                        if isinstance(provenance, dict)
+                        else None
+                    )
+                score = sources.count("execution")
+                source = (
+                    "execution"
+                    if score == 2
+                    else (
+                        "mixed"
+                        if score
+                        else (
+                            "embedded-parameters"
+                            if sources == ["embedded-parameters"] * 2
+                            else "graph-inferred"
+                        )
+                    )
+                )
+                candidates.append((score, positive, negative, source))
+            context = None
+            if candidates and pair:
+                best_score = max(item[0] for item in candidates)
+                contexts = {item[1:] for item in candidates if item[0] == best_score}
+                if len(contexts) == 1:
+                    positive, negative, source = next(iter(contexts))
+                    context = {
+                        "positive_text": positive,
+                        "negative_text": negative,
+                        "text_source": source,
+                    }
            tags = self._get_prompt_tags(conn, row[0])
            try:
                workflow = json.loads(row[5]) if row[5] else None
@@ -2010,6 +2110,8 @@ class PromptDatabase:
            except (json.JSONDecodeError, TypeError):
                prompt_meta = None
            return {
+                "pair": pair,
+                "pair_context": context,
                "prompt_id": row[0],
                "text": row[1],
                "category": row[2],
@@ -2050,11 +2152,15 @@ class PromptDatabase:

        Returns number of prompts removed.
        """
-        with self.model.get_connection() as conn:
+        with self._pair_transaction() as conn:
            cursor = conn.execute("""
                SELECT p.id FROM prompts p
                LEFT JOIN generated_images gi ON p.id = gi.prompt_id
                WHERE gi.prompt_id IS NULL
+                AND NOT EXISTS (
+                    SELECT 1 FROM prompt_pairs pp
+                    WHERE pp.positive_prompt_id = p.id OR pp.negative_prompt_id = p.id
+                )
                AND NOT EXISTS (
                    SELECT 1 FROM prompt_tags pt
                    JOIN tags t ON pt.tag_id = t.id
@@ -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,
+        }
@@ -0,0 +1,387 @@
+"""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:
+        prompt_hash = generate_prompt_hash(text)
+        conn.execute(
+            "INSERT INTO prompts(text, hash, created_at, updated_at) VALUES(?,?,?,?) "
+            "ON CONFLICT(hash) DO NOTHING",
+            (text.strip(), prompt_hash, utc_now_iso(), utc_now_iso()),
+        )
+        return conn.execute(
+            "SELECT id FROM prompts WHERE hash = ?", (prompt_hash,)
+        ).fetchone()["id"]
+
+    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")
+        tags = _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
+        )
+        existing = conn.execute(
+            "SELECT id FROM prompt_pairs WHERE positive_prompt_id = ? "
+            "AND negative_prompt_id IS ?",
+            (positive, negative),
+        ).fetchone()
+        if existing:
+            return existing["id"]
+        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, negative, title, category, rating, notes, now, now),
+        ).lastrowid
+        if tags:
+            self._sync_pair_tags(conn, pair_id, tags)
+        return pair_id
+
+    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
+        ]
@@ -0,0 +1,83 @@
+# 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).
+- [#208](https://github.com/ComfyAssets/ComfyUI_PromptManager/issues/208) adds the native Prompt Library: 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) now records new job outcomes; prompt contexts, output associations and the execution timeline remain pending under #209.
+- Prompt execution contexts and timeline, tagging/review and System consolidation remain subsequent slices under #196; #209 now records durable job-level outcomes.
+
+## 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.
@@ -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.
@@ -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>
@@ -0,0 +1,62 @@
+# Durable Prompt History
+
+Status: the durable job ledger is implemented as the first slice of
+[#209](https://github.com/ComfyAssets/ComfyUI_PromptManager/issues/209). Prompt
+contexts, output 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
+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.
+
+## Remaining history work
+
+Design bounded execution contexts with nullable recipe references, immutable
+raw/effective positive, negative and composition snapshots and explicit
+inferred/recorded/unknown provenance. Cached nodes must not inherit unrelated
+recent prompt evidence. Define deletion and restore behavior before connecting
+saved recipes to historical snapshots.
+
+Add exact output references and a deduplicated run/image relationship without
+claiming cached outputs were newly created. Preserve existing image provenance
+and same-path replacement protection. Only after those semantics are tested can
+the timeline honestly show prompt-level execution chronology, batched outputs
+and repeat-run branches.
@@ -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.
@@ -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.
@@ -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.
@@ -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.
@@ -0,0 +1,71 @@
+# Shared ComfyAssets UI adoption
+
+Status: approved direction; toolkit/testing foundation in progress.
+Product UI brainstorming comes before node changes.
+
+Tracking: [PromptManager #186](https://github.com/ComfyAssets/ComfyUI_PromptManager/issues/186)
+and [KikoTools #81](https://github.com/ComfyAssets/ComfyUI-KikoTools/issues/81).
+
+KikoTools is the validated model. The
+[ComfyAssets Toolkit](https://github.com/ComfyAssets/ComfyAssets-Toolkit) is the
+shared source; PromptManager proves the package boundaries before a later
+KikoTools adoption PR. Other custom nodes can become additional consumers.
+
+## Approved sequence
+
+1. Extract reusable controls and test infrastructure from the working KikoTools
+   reference into the toolkit. Keep routes, node schemas and controllers local.
+2. Bring PromptManager's automated tests up to parity using the pinned toolkit.
+3. Brainstorm PromptManager node layouts, controls and behavior with the maintainer
+   before changing its production UI. Select one bounded first consumer together.
+4. Validate shared distribution, themes, cleanup and compatibility through that
+   consumer, then migrate incrementally. Test both packs together as appropriate.
+5. Backport toolkit adoption to KikoTools after PromptManager proves it, preserving
+   the existing KikoTools appearance, behavior and workflow compatibility.
+
+KikoTools is not being migrated again during this phase. The original
+[foundation mockup](https://github.com/ComfyAssets/ComfyUI-KikoTools/pull/82)
+records the approved visual direction and inspiration.
+
+## PromptManager requirements
+
+- Preserve existing node names, inputs/outputs, API routes, saved workflows,
+  gallery behavior, and existing test/coverage gates.
+- Use shared control factories, editor slots, settings/help/window behavior and
+  the ComfyUI adapter where applicable. Keep PromptManager controllers and HTTP
+  behavior in this repository.
+- Both projects must look identical: one shared set of controls, typography,
+  spacing, colors and interaction states. Follow ComfyUI light/dark mode with
+  our own styling, a blue primary accent and strong red errors/destructive actions.
+  Use compact buttons, pill switches, segmented selections and responsive rows.
+  Review the pilot mockup before production styling. Existing `--pm-*` tokens
+  are migration context, not an approved design to retain.
+- Theme standalone pages explicitly: `window.open` pages do not inherit the
+  host's CSS variables. Test the agreed palette policy, live changes, fallbacks
+  and persistence, including errors when host settings are unavailable.
+- Ship a pinned, reproducible browser distribution. Installation must work
+  without npm or KikoTools, including when both packs ship different core versions.
+- Attribute [Pixaroma](https://github.com/pixaroma/ComfyUI-Pixaroma) as architectural
+  inspiration. Do not copy/vendor his code, styles, or assets.
+
+## First migration
+
+With the maintainer, inventory the PromptManager nodes and gallery dialogs and
+choose one bounded first UI consumer after brainstorming. Record its current behavior with tests before changing it.
+Verify keyboard opening/closing, focus restoration, accessible labels, confirm
+and cancel behavior, network failures, and light/dark appearances. Keep the
+existing gallery and destructive-action browser workflows passing.
+
+Build shared-source extraction and distribution checks before importing the core.
+Test both packs alone and together, scoped styling, isolated settings/storage,
+and independent cleanup. ComfyUI-facing changes require the real-host Classic,
+Nodes 2.0 and live-switching matrix from the foundation proposal. Standalone
+dialogs additionally run through PromptManager's existing browser harness.
+
+## Delivery
+
+Use small issue-linked PRs and TDD: verify each new behavioral test fails before
+implementing it, then run relevant suites and required CI checks. Include visual
+evidence for appearance changes. Migrate additional nodes/dialogs/editors only after
+the first consumer validates the package boundaries; do not rewrite all pages
+or change backend schemas as part of this initiative.
@@ -6,6 +6,11 @@ with persistent prompt storage and search capabilities.
 import time
 from typing import Any, Dict, Optional, Tuple

+if __package__:
+    from .utils.execution_context_capture import record_runtime_context
+else:
+    from utils.execution_context_capture import record_runtime_context
+
 try:
    from comfy.comfy_types import IO, ComfyNodeABC, InputTypeDict
 except ImportError:
@@ -256,7 +261,11 @@ class PromptManager(PromptManagerBase, ComfyNodeABC):
        )

        self.logger.info(f"CLIP encoding completed, text: {repr(encoding_text)[:80]}")
-        return (conditioning, encoding_text)
+        result = (conditioning, encoding_text)
+        record_runtime_context(
+            type(self).__name__, text, prepend_text, append_text, encoding_text
+        )
+        return result

    @classmethod
    def IS_CHANGED(
@@ -281,3 +290,22 @@ class PromptManager(PromptManagerBase, ComfyNodeABC):

        combined = f"{text}|{prepend_text}|{append_text}|{category}|{tags}"
        return hashlib.sha256(combined.encode()).hexdigest()
+
+
+class PromptManagerFocused(PromptManager):
+    """The editor-first UI uses the existing encoding and storage contract."""
+
+    @classmethod
+    def INPUT_TYPES(cls) -> InputTypeDict:
+        inputs = super().INPUT_TYPES()
+        for name in ("prepend_text", "append_text"):
+            kind, options = inputs["optional"][name]
+            inputs["optional"][name] = (
+                kind,
+                {**options, "default": "", "forceInput": True},
+            )
+        return inputs
+
+
+class PromptManagerSplit(PromptManagerFocused):
+    """The persistent library layout shares Focused encoding and storage."""
@@ -0,0 +1,243 @@
+"""Paired positive and negative conditioning with reusable recipe storage."""
+
+import hashlib
+import json
+import time
+from typing import Any, Dict, Optional, Tuple
+
+if __package__:
+    from .utils.execution_context_capture import record_runtime_context
+else:
+    from utils.execution_context_capture import record_runtime_context
+
+try:
+    from .prompt_manager import IO, InputTypeDict, PromptManagerFocused
+    from .utils.hashing import generate_prompt_hash
+    from .utils.pair_execution import record_pair_execution
+    from .utils.prompt_graph import intrinsic_pair_nodes, run_prompt_components
+    from .utils.prompt_pair_resolver import resolve_text_input
+    from .utils.usage_tracking import (
+        PENDING_FIRST_USE,
+        PENDING_PAIR_FIRST_USE,
+        pair_run_key,
+    )
+except ImportError:
+    from prompt_manager import IO, InputTypeDict, PromptManagerFocused
+    from utils.hashing import generate_prompt_hash
+    from utils.pair_execution import record_pair_execution
+    from utils.prompt_graph import intrinsic_pair_nodes, run_prompt_components
+    from utils.prompt_pair_resolver import resolve_text_input
+    from utils.usage_tracking import (
+        PENDING_FIRST_USE,
+        PENDING_PAIR_FIRST_USE,
+        pair_run_key,
+    )
+
+
+class PromptManagerPair(PromptManagerFocused):
+    """Keep raw editors as one recipe while encoding their effective texts."""
+
+    @classmethod
+    def INPUT_TYPES(cls) -> InputTypeDict:
+        inputs = super().INPUT_TYPES()
+        inputs["required"] = {
+            "text": inputs["required"]["text"],
+            "negative_text": (
+                IO.STRING,
+                {"default": "", "multiline": True, "dynamicPrompts": True},
+            ),
+            "clip": inputs["required"]["clip"],
+        }
+        inputs["hidden"]["extra_pnginfo"] = "EXTRA_PNGINFO"
+        return inputs
+
+    RETURN_TYPES = (IO.CONDITIONING, IO.CONDITIONING, IO.STRING, IO.STRING)
+    RETURN_NAMES = ("positive", "negative", "positive_text", "negative_text")
+    OUTPUT_TOOLTIPS = (
+        "Positive conditioning with composition and optional LoRA trigger words.",
+        "Negative conditioning from the negative editor, including empty text.",
+        "The effective positive text that was encoded.",
+        "The unchanged negative text that was encoded.",
+    )
+    DESCRIPTION = "Encode and save a reusable positive/negative prompt pair."
+
+    def encode_prompt(
+        self,
+        clip: Any,
+        text: str,
+        negative_text: str = "",
+        category: str = "",
+        tags: str = "",
+        search_text: str = "",
+        prepend_text: str = "",
+        append_text: str = "",
+        prompt: Optional[Dict[str, Any]] = None,
+        unique_id: Optional[str] = None,
+        extra_pnginfo: Optional[Dict[str, Any]] = None,
+    ) -> Tuple[Any, Any, str, str]:
+        """Encode both sides once; database and tracking failures never block CLIP."""
+        if clip is None:
+            raise RuntimeError("ERROR: clip input is invalid: None")
+        if not isinstance(text, str) or not isinstance(negative_text, str):
+            raise ValueError("Positive and negative prompt text must be strings")
+        positive = self._inject_lora_trigger_words(
+            " ".join(
+                part
+                for part in (prepend_text.strip(), text, append_text.strip())
+                if part
+            )
+        )
+        positive_conditioning = clip.encode_from_tokens_scheduled(
+            clip.tokenize(positive)
+        )
+        negative_conditioning = clip.encode_from_tokens_scheduled(
+            clip.tokenize(negative_text)
+        )
+        record_pair_execution(
+            extra_pnginfo,
+            unique_id,
+            text=text,
+            negative_text=negative_text,
+            positive_text=positive,
+            negative_effective_text=negative_text,
+            prepend_text=prepend_text.strip(),
+            append_text=append_text.strip(),
+        )
+        pair_id = prompt_id = None
+        known = isinstance(prompt, dict) and unique_id is not None
+        node_id = str(unique_id)
+        used_fields = (
+            {
+                field
+                for source, field in run_prompt_components(prompt)
+                if source == node_id
+            }
+            if known
+            else {"text"}
+        )
+        is_positive = "text" in used_fields
+        pair_used = node_id in intrinsic_pair_nodes(prompt) if known else True
+        try:
+            pair_id = self.db.save_prompt_pair(
+                text,
+                negative_text,
+                category=category.strip() or None,
+                tags=self._parse_tags(tags),
+            )
+            pair = self.db.get_prompt_pair(pair_id)
+            prompt_id = pair["positive_prompt_id"]
+            resolved_positive = (
+                resolve_text_input(prompt, node_id, "text") if known else None
+            )
+            resolved_negative = (
+                resolve_text_input(prompt, node_id, "negative_text") if known else None
+            )
+            positive_seen = resolved_positive is not None and generate_prompt_hash(
+                resolved_positive
+            ) == generate_prompt_hash(text)
+            pair_seen = (
+                positive_seen
+                and resolved_negative is not None
+                and generate_prompt_hash(resolved_negative)
+                == generate_prompt_hash(negative_text)
+            )
+            positive_pending = getattr(self, "pending_first_use", PENDING_FIRST_USE)
+            pair_pending = getattr(
+                self, "pending_pair_first_use", PENDING_PAIR_FIRST_USE
+            )
+            counted = set()
+            for field, component_text, resolved_text, component_id in (
+                ("text", text, resolved_positive, prompt_id),
+                (
+                    "negative_text",
+                    negative_text,
+                    resolved_negative,
+                    pair["negative_prompt_id"],
+                ),
+            ):
+                component_hash = generate_prompt_hash(component_text)
+                if (
+                    field not in used_fields
+                    or not component_id
+                    or component_hash in counted
+                ):
+                    continue
+                counted.add(component_hash)
+                seen_at_queue = (
+                    resolved_text is not None
+                    and generate_prompt_hash(resolved_text) == component_hash
+                )
+                runs = positive_pending.consume(component_hash) + int(not seen_at_queue)
+                if runs:
+                    self.db.record_prompt_use(component_id, times=runs)
+            if pair_used:
+                runs = pair_pending.consume(pair_run_key(text, negative_text)) + int(
+                    not pair_seen
+                )
+                if runs:
+                    self.db.record_prompt_pair_use(pair_id, times=runs)
+            if is_positive:
+                self.prompt_tracker.set_current_prompt(
+                    prompt_text=positive,
+                    additional_data={
+                        "prompt_id": prompt_id,
+                        "pair_id": pair_id if pair_used else None,
+                        "positive_text": positive,
+                        "negative_text": negative_text,
+                        "positive_storage_text": text.strip(),
+                        "negative_storage_text": negative_text.strip(),
+                        "prepend_text": prepend_text.strip(),
+                        "append_text": append_text.strip(),
+                        "final_text": positive,
+                        "category": category.strip() or None,
+                        "tags": self._parse_tags(tags) or [],
+                    },
+                    push_to_queue=not positive_seen or (pair_used and not pair_seen),
+                )
+        except Exception as error:
+            self.logger.warning(f"Could not save or track paired prompt: {error}")
+        try:
+            self.comfyui_integration.register_prompt(
+                f"promptmanager_pair_{time.time_ns()}",
+                positive,
+                {
+                    "prompt_id": prompt_id,
+                    "pair_id": pair_id if pair_used else None,
+                    "negative_text": negative_text,
+                    "category": category.strip() or None,
+                },
+            )
+        except Exception as error:
+            self.logger.warning(f"Could not register paired metadata: {error}")
+        result = positive_conditioning, negative_conditioning, positive, negative_text
+        record_runtime_context(
+            type(self).__name__,
+            text,
+            prepend_text,
+            append_text,
+            positive,
+            role="positive",
+        )
+        record_runtime_context(
+            type(self).__name__, negative_text, "", "", negative_text, role="negative"
+        )
+        return result
+
+    @classmethod
+    def IS_CHANGED(
+        cls,
+        clip: Any,
+        text: str = "",
+        negative_text: str = "",
+        category: str = "",
+        tags: str = "",
+        search_text: str = "",
+        prepend_text: str = "",
+        append_text: str = "",
+        **kwargs: Any,
+    ) -> str:
+        """Either editor, composition or pair metadata invalidates the node cache."""
+        serialized = json.dumps(
+            [text, negative_text, prepend_text, append_text, category, tags]
+        )
+        return hashlib.sha256(serialized.encode("utf-8")).hexdigest()
@@ -6,6 +6,11 @@ without CLIP encoding, while maintaining all database and search features.
 import time
 from typing import Any, Dict, Optional, Tuple

+if __package__:
+    from .utils.execution_context_capture import record_runtime_context
+else:
+    from utils.execution_context_capture import record_runtime_context
+
 try:
    from comfy.comfy_types import IO, ComfyNodeABC, InputTypeDict
 except ImportError:
@@ -227,7 +232,11 @@ class PromptManagerText(PromptManagerBase, ComfyNodeABC):
        )

        self.logger.debug(f"Text processing completed: {final_text[:100]}...")
-        return (final_text,)
+        result = (final_text,)
+        record_runtime_context(
+            type(self).__name__, text, prepend_text, append_text, final_text
+        )
+        return result

    @classmethod
    def IS_CHANGED(
@@ -22,6 +22,8 @@ from aiohttp import web
 from PIL import Image

 from .prompts import PromptRoutesMixin
+from .pairs import PairRoutesMixin
+from .library import LibraryRoutesMixin
 from .images import ImageRoutesMixin
 from .admin import AdminRoutesMixin
 from .logging_routes import LoggingRoutesMixin
@@ -84,6 +86,7 @@ def _public_error(exc: BaseException) -> str:
 # doubles as the allow-list.
 STATIC_MIME_TYPES = {
    ".js": "application/javascript",
+    ".mjs": "application/javascript",
    ".css": "text/css",
    ".map": "application/json",
    ".woff": "font/woff",
@@ -240,6 +243,8 @@ async def _gzip_middleware(request, handler):


 class PromptManagerAPI(
+    PairRoutesMixin,
+    LibraryRoutesMixin,
    PromptRoutesMixin,
    ImageRoutesMixin,
    AdminRoutesMixin,
@@ -262,12 +267,12 @@ class PromptManagerAPI(
        db (PromptDatabase): Database connection and operations handler
    """

-    def __init__(self):
-        """Initialize the PromptManager API with database connection and cleanup."""
+    def __init__(self, db: PromptDatabase | None = None) -> None:
+        """Initialize the API, using an injected database before any cleanup."""
        self.logger = get_logger("prompt_manager.api")
        self.logger.info("Initializing PromptManager API")

-        self.db = PromptDatabase()
+        self.db = db if db is not None else PromptDatabase()
        self._cached_output_dir = None  # Lazy-cached by _find_comfyui_output_dir()
        self._html_cache = {}  # Cached HTML file contents keyed by path
        self._gallery_cache = {}  # dict: path_str -> (files, timestamp)
@@ -361,6 +366,10 @@ class PromptManagerAPI(

        # ── Web UI serving routes ─────────────────────────────────────

+        @routes.get("/prompt_manager/workspace")
+        async def serve_workspace_ui(request: web.Request) -> web.Response:
+            return self._serve_html_page("workspace.html", "Workspace")
+
        @routes.get("/prompt_manager/web")
        async def serve_web_ui(request):
            return self._serve_html_page("index.html", "Web UI", cache=False)
@@ -391,6 +400,8 @@ class PromptManagerAPI(

        # ── Register domain-specific routes from mixins ───────────────

+        self._register_pair_routes(routes)
+        self._register_library_routes(routes)
        self._register_prompt_routes(routes)
        self._register_image_routes(routes)
        self._register_admin_routes(routes)
@@ -26,8 +26,12 @@ except ImportError:

 try:
    from ...database.operations import PromptDatabase
+    from ...database.pair_ingestion import ingest_resolved_pair
+    from ...utils.prompt_pair_resolver import resolve_prompt_pair
 except ImportError:
    from database.operations import PromptDatabase
+    from database.pair_ingestion import ingest_resolved_pair
+    from utils.prompt_pair_resolver import resolve_prompt_pair


 try:
@@ -40,6 +44,20 @@ VIDEO_EXTENSIONS = (".mp4", ".avi", ".mov", ".mkv", ".webm", ".m4v", ".wmv")
 SCAN_BATCH_SIZE = 50  # Files per executor call during the output scan
 MAX_SAFETY_BACKUPS = 10  # '<db>.backup_*' files kept next to the live database
 MAX_SCAN_FILES = 50_000  # Upper bound on media files one scan will look at
+SCAN_COUNT_KEYS = (
+    "processed",
+    "found",
+    "added",
+    "linked",
+    "enriched",
+    "unchanged",
+    "ambiguous",
+    "unresolved",
+    "conflict",
+    "failed",
+    "pairs_added",
+    "pair_links_added",
+)
 MAX_SCAN_DEPTH = 12  # Directory levels below an output root that are scanned
 # Files the duplicate cleaner may delete: everything the scan reports, plus
 # the common single-f TIFF spelling. Anything else inside the output
@@ -359,6 +377,10 @@ class AdminRoutesMixin:
        async def scan_images_route(request):
            return await self.scan_images(request)

+        @routes.post("/prompt_manager/scan/cancel")
+        async def cancel_scan_route(request):
+            return await self.cancel_scan(request)
+
        @routes.get("/prompt_manager/scan/status")
        async def scan_status_route(request):
            return await self.scan_status(request)
@@ -1280,6 +1302,14 @@ class AdminRoutesMixin:
            job = self._start_scan_job()
        return await self._stream_scan_job(request, job)

+    async def cancel_scan(self, request):
+        """Stop between file transactions; an in-flight extraction can finish."""
+        job = self._scan_job
+        requested = bool(job and job.running)
+        if requested:
+            job.cancelled = True
+        return web.json_response({"success": True, "cancellation_requested": requested})
+
    async def scan_status(self, request):
        """Whether an output scan is running, and its latest progress event."""
        job = self._scan_job
@@ -1351,7 +1381,7 @@ class AdminRoutesMixin:
            media_files = await self._run_in_executor(
                self._collect_output_media_sync, output_dirs
            )
-            counts = {"processed": 0, "found": 0, "added": 0, "linked": 0}
+            counts = dict.fromkeys(SCAN_COUNT_KEYS, 0)
            if media_files:
                yield self._scan_progress(
                    5, f"Found {len(media_files)} media files to process...", counts
@@ -1364,7 +1394,9 @@ class AdminRoutesMixin:
                f"found={counts['found']}, new_prompts_added={counts['added']}, "
                f"images_linked_to_existing={counts['linked']}"
            )
-            yield {"type": "complete", **counts, "directories": dir_names}
+            job = self._scan_job
+            final_type = "cancelled" if job and job.cancelled else "complete"
+            yield {"type": final_type, **counts, "directories": dir_names}

        except Exception:
            self.logger.exception("Scan error")
@@ -1380,6 +1412,8 @@ class AdminRoutesMixin:
        """
        total = len(media_files)
        for batch_start in range(0, total, SCAN_BATCH_SIZE):
+            if self._scan_job and self._scan_job.cancelled:
+                return
            batch = media_files[batch_start : batch_start + SCAN_BATCH_SIZE]
            batch_results = await self._run_in_executor(
                self._extract_batch_metadata_sync, batch
@@ -1405,7 +1439,7 @@ class AdminRoutesMixin:
            "progress": progress,
            "status": status,
            "processed": counts.get("processed", 0),
-            "found": counts.get("found", 0),
+            **{key: counts.get(key, 0) for key in SCAN_COUNT_KEYS},
        }

    def _collect_output_media_sync(self, output_dirs):
@@ -1431,25 +1465,64 @@ class AdminRoutesMixin:

    def _ingest_scan_batch_sync(self, batch_results):
        """Store the prompts found in a metadata batch (blocking DB work)."""
-        counts = {"processed": 0, "found": 0, "added": 0, "linked": 0}
+        counts = dict.fromkeys(SCAN_COUNT_KEYS, 0)
        for media_file, metadata in batch_results:
+            if self._scan_job and self._scan_job.cancelled:
+                break
            counts["processed"] += 1
            try:
-                outcome = self._ingest_scanned_file(media_file, metadata)
+                outcome = self._ingest_scanned_file(media_file, metadata, detailed=True)
            except Exception as e:
                self.logger.error(f"Error processing {media_file.name}: {e}")
+                counts["failed"] += 1
                continue
            if outcome is None:
                continue
            counts["found"] += 1
-            if outcome in ("added", "linked"):
-                counts[outcome] += 1
+            status = outcome["status"] if isinstance(outcome, dict) else outcome
+            if status in (
+                "added",
+                "linked",
+                "enriched",
+                "unchanged",
+                "ambiguous",
+                "unresolved",
+                "conflict",
+                "failed",
+            ):
+                counts[status] += 1
+            if isinstance(outcome, dict):
+                counts["pairs_added"] += outcome.get("pairs_added", 0)
+                counts["pair_links_added"] += outcome.get("links_added", 0)
        return counts

-    def _ingest_scanned_file(self, media_file, metadata):
-        """Save or link one scanned file; returns 'added', 'linked', 'found' or None."""
+    def _ingest_scanned_file(self, media_file, metadata, detailed=False):
+        """Enrich exact pairs; retain legacy positive-only import for absent pairs."""
        if not metadata:
            return None
+        canonical = dict(metadata)
+        for key in ("prompt", "workflow"):
+            if isinstance(canonical.get(key), str):
+                try:
+                    canonical[key] = json.loads(canonical[key])
+                except (ValueError, TypeError):
+                    # Preserve malformed metadata for conservative resolver rejection.
+                    pass
+        if not canonical.get("prompt") and not canonical.get("parameters"):
+            return None
+        resolution = resolve_prompt_pair(canonical)
+        if resolution["state"] == "resolved":
+            result = ingest_resolved_pair(
+                self.db, str(media_file), canonical, resolution
+            )
+            return result if detailed else result["status"]
+        if resolution["state"] == "ambiguous":
+            return "ambiguous"
+        if resolution["state"] == "unresolved":
+            if canonical.get("prompt") or not isinstance(
+                canonical.get("parameters"), str
+            ):
+                return resolution["state"]
        parsed = self._parse_comfyui_prompt(metadata)
        if not (parsed.get("prompt") or parsed.get("parameters")):
            return None
@@ -1465,13 +1538,15 @@ class AdminRoutesMixin:
        existing = self.db.get_prompt_by_hash(prompt_hash)
        if existing:
            try:
-                self.db.link_image_to_prompt(existing["id"], str(media_file))
+                image_id = self.db.link_image_to_prompt(
+                    existing["id"], str(media_file), metadata=canonical
+                )
            except Exception as e:
                self.logger.error(
                    f"Failed to link {media_file.name} to existing prompt: {e}"
                )
                return "found"
-            return "linked"
+            return "linked" if image_id else "unchanged"

        prompt_id = self.db.save_prompt(
            prompt_text,
@@ -1484,7 +1559,7 @@ class AdminRoutesMixin:
        if not prompt_id:
            return "found"
        try:
-            self.db.link_image_to_prompt(prompt_id, str(media_file))
+            self.db.link_image_to_prompt(prompt_id, str(media_file), metadata=canonical)
        except Exception as e:
            self.logger.error(f"Failed to link {media_file.name} to new prompt: {e}")
        return "added"
@@ -1,7 +1,10 @@
 """Image and gallery API routes for PromptManager."""

 import asyncio
+import base64
+import binascii
 import hashlib
+import hmac
 import json
 import os
 import re
@@ -19,8 +22,18 @@ from .prompts import safe_error_message as _safe_error

 try:
    from ...utils.parallel import map_parallel
+    from ...utils.prompt_inspector import (
+        MetadataReadError,
+        inspect_prompt_roles,
+        read_inspection_metadata,
+    )
 except ImportError:
    from utils.parallel import map_parallel
+    from utils.prompt_inspector import (
+        MetadataReadError,
+        inspect_prompt_roles,
+        read_inspection_metadata,
+    )

 # Only these file types are ever served by the image routes, regardless of
 # what sits inside an allowed directory (defence in depth for the gallery).
@@ -228,6 +241,70 @@ MAX_JSON_BODY_BYTES = 1024 * 1024
 _READ_CHUNK = 64 * 1024


+_WORKSPACE_CURSOR_SECRET = os.urandom(32)
+
+
+def _workspace_root(query: dict, roots: list[Path]) -> int | None:
+    value = query.get("root")
+    if value is None:
+        return None
+    try:
+        index = int(value)
+    except ValueError:
+        raise ValueError("Invalid gallery root") from None
+    if not 0 <= index < len(roots):
+        raise ValueError("Invalid gallery root")
+    return index
+
+
+def _workspace_key(entry: tuple, order: str) -> tuple:
+    path, mtime, root, index = entry
+    relative = path.relative_to(root).as_posix()
+    tie = hashlib.sha256(relative.encode("utf-8")).hexdigest()
+    name = path.name.casefold()
+    if order in ("name", "name-desc"):
+        return (name, tie, index, mtime)
+    return ((-mtime if order == "newest" else mtime), name, tie, index)
+
+
+def _workspace_cursor(key: tuple, scope: str) -> str:
+    payload = json.dumps(
+        {"v": 1, "scope": scope, "key": key}, separators=(",", ":"), ensure_ascii=False
+    ).encode()
+    signature = hmac.new(_WORKSPACE_CURSOR_SECRET, payload, hashlib.sha256).digest()
+    return base64.urlsafe_b64encode(signature + payload).decode().rstrip("=")
+
+
+def _workspace_cursor_key(token: str, scope: str, order: str) -> tuple:
+    try:
+        if len(token) > 4096:
+            raise ValueError()
+        raw = base64.b64decode(
+            token + "=" * (-len(token) % 4), altchars=b"-_", validate=True
+        )
+        signature, payload = raw[:32], raw[32:]
+        expected = hmac.new(_WORKSPACE_CURSOR_SECRET, payload, hashlib.sha256).digest()
+        if not hmac.compare_digest(signature, expected):
+            raise ValueError()
+        data = json.loads(payload)
+        key = data["key"]
+        if data["v"] != 1 or data["scope"] != scope or not isinstance(key, list):
+            raise ValueError()
+        types = (
+            (str, str, int, (float, int))
+            if order in ("name", "name-desc")
+            else ((float, int), str, str, int)
+        )
+        if len(key) != len(types) or not all(
+            type(value) in (kind if isinstance(kind, tuple) else (kind,))
+            for value, kind in zip(key, types)
+        ):
+            raise ValueError()
+        return tuple(key)
+    except (ValueError, KeyError, TypeError, UnicodeError, binascii.Error):
+        raise ValueError("Invalid gallery cursor; refresh the gallery") from None
+
+
 class _BodyTooLarge(Exception):
    """Raised when a request body exceeds MAX_JSON_BODY_BYTES."""

@@ -381,6 +458,10 @@ class ImageRoutesMixin:
        async def get_image_prompt_route(request):
            return await self.get_image_prompt(request)

+        @routes.get("/prompt_manager/images/inspect/{image_path:.*}")
+        async def inspect_image_route(request):
+            return await self.inspect_image(request)
+
        @routes.delete("/prompt_manager/images/{image_id}")
        async def delete_image_route(request):
            return await self.delete_image(request)
@@ -594,6 +675,8 @@ class ImageRoutesMixin:
        """Get all images from ComfyUI output folder(s)."""
        try:
            output_dirs = self._get_all_output_dirs()
+            if request.query.get("mode") == "workspace":
+                return await self._get_workspace_images(request, list(output_dirs))
            if not output_dirs:
                return web.json_response(
                    {
@@ -629,6 +712,119 @@ class ImageRoutesMixin:
                {"success": False, "error": _safe_error(e)}, status=500
            )

+    async def _get_workspace_images(
+        self, request, output_dirs: list[Path]
+    ) -> web.Response:
+        try:
+            limit, _offset = parse_page_params(request.query, default_limit=100)
+        except ValueError:
+            return bad_request("limit and offset must be integers")
+        try:
+            root_index = _workspace_root(request.query, output_dirs)
+            query = request.query.get("q", "").strip().casefold()
+            media = request.query.get("media", "all")
+            order = request.query.get("sort", "newest")
+            folder = request.query.get("subfolder", "").strip().replace("\\", "/")
+            if (
+                len(query) > 500
+                or media not in ("all", "image", "video")
+                or order not in ("newest", "oldest", "name", "name-desc")
+            ):
+                raise ValueError("Invalid gallery filters")
+            if folder and (_has_traversal(folder) or len(folder) > 1024):
+                raise ValueError("Invalid gallery folder")
+            scope = hashlib.sha256(
+                json.dumps(
+                    [
+                        query,
+                        media,
+                        order,
+                        folder,
+                        root_index,
+                        [str(root) for root in output_dirs],
+                    ]
+                ).encode()
+            ).hexdigest()
+            token = request.query.get("cursor")
+            after = _workspace_cursor_key(token, scope, order) if token else None
+        except ValueError as exc:
+            return bad_request(str(exc))
+
+        listings = [
+            (index, root, await self._get_gallery_files(root))
+            for index, root in enumerate(output_dirs)
+            if root_index is None or index == root_index
+        ]
+        page, total, has_more, next_cursor = await self._run_in_executor(
+            self._select_workspace_page,
+            listings,
+            folder,
+            query,
+            media,
+            order,
+            after,
+            limit,
+            scope,
+        )
+        images = await self._run_in_executor(self._format_workspace_page, page)
+        return web.json_response(
+            {
+                "success": True,
+                "images": images,
+                "total": total,
+                "limit": limit,
+                "has_more": has_more,
+                "next_cursor": next_cursor,
+            }
+        )
+
+    @staticmethod
+    def _select_workspace_page(
+        listings: list[tuple],
+        folder: str,
+        query: str,
+        media: str,
+        order: str,
+        after: tuple | None,
+        limit: int,
+        scope: str,
+    ) -> tuple:
+        entries = []
+        for index, root, files in listings:
+            for path, mtime in files:
+                relative = path.relative_to(root)
+                if folder and not _in_subfolder(
+                    str(relative.parent), folder.replace("/", os.sep)
+                ):
+                    continue
+                if query and query not in relative.as_posix().casefold():
+                    continue
+                kind = "video" if path.suffix.lower() in VIDEO_EXTENSIONS else "image"
+                if media != "all" and media != kind:
+                    continue
+                entries.append((path, mtime, root, index))
+        total = len(entries)
+        reverse = order == "name-desc"
+        entries.sort(key=lambda entry: _workspace_key(entry, order), reverse=reverse)
+        if after is not None:
+            entries = [
+                entry
+                for entry in entries
+                if (
+                    _workspace_key(entry, order) < after
+                    if reverse
+                    else _workspace_key(entry, order) > after
+                )
+            ]
+        page = entries[:limit]
+        has_more = len(entries) > limit
+        next_cursor = (
+            _workspace_cursor(_workspace_key(page[-1], order), scope)
+            if has_more
+            else None
+        )
+        return page, total, has_more, next_cursor
+
    async def serve_image(self, request):
        """Serve the actual image file using streamed FileResponse."""
        try:
@@ -775,7 +971,20 @@ class ImageRoutesMixin:
    async def get_gallery_subfolders(self, request):
        """Get distinct subfolders from gallery output directories."""
        try:
-            output_dirs = self._get_all_output_dirs()
+            output_dirs = list(self._get_all_output_dirs())
+            workspace = request.query.get("mode") == "workspace"
+            roots = None
+            if workspace:
+                try:
+                    selected = _workspace_root(request.query, output_dirs)
+                except ValueError as exc:
+                    return bad_request(str(exc))
+                roots = [
+                    {"index": index, "label": _public_path(root)}
+                    for index, root in enumerate(output_dirs)
+                ]
+                if selected is not None:
+                    output_dirs = [output_dirs[selected]]
            subfolders = set()
            for output_path in output_dirs:
                files = await self._get_gallery_files(output_path)
@@ -795,9 +1004,10 @@ class ImageRoutesMixin:
                        ancestors.add("/".join(parts[:i]))
                subfolders.update(ancestors)

-            return web.json_response(
-                {"success": True, "subfolders": sorted(subfolders)}
-            )
+            payload = {"success": True, "subfolders": sorted(subfolders)}
+            if workspace:
+                payload["roots"] = roots
+            return web.json_response(payload)
        except Exception as e:
            self.logger.error(f"Subfolders error: {e}")
            return web.json_response(
@@ -1432,6 +1642,65 @@ class ImageRoutesMixin:
                {"success": False, "error": _safe_error(e)}, status=500
            )

+    async def inspect_image(self, request):
+        """Read prompt evidence for one explicitly rooted physical media file."""
+        roots = list(self._get_all_output_dirs())
+        try:
+            index = _workspace_root(request.query, roots)
+            if index is None:
+                raise ValueError("Gallery root is required")
+        except ValueError as error:
+            return bad_request(str(error))
+        relative = request.match_info.get("image_path", "")
+        if (
+            _has_traversal(relative)
+            or "\0" in relative
+            or re.match(r"^[a-zA-Z]:", relative)
+        ):
+            return _forbidden()
+        path = Path(roots[index]) / relative
+        denied = _media_access_error(path, [roots[index]])
+        if denied is not None:
+            return denied
+        if not path.is_file():
+            return web.json_response(
+                {"success": False, "error": "Image file not found"}, status=404
+            )
+        try:
+
+            def inspect() -> dict:
+                try:
+                    metadata = read_inspection_metadata(path)
+                except MetadataReadError:
+                    return {
+                        role: {"status": "unresolved", "text": None, "source": None}
+                        for role in ("positive", "negative")
+                    }
+                saved = (
+                    self.db.get_image_prompt_info(str(path), False)
+                    if not metadata
+                    else None
+                )
+                return inspect_prompt_roles(metadata, saved)
+
+            prompts = await self._run_in_executor(inspect)
+            return web.json_response(
+                {
+                    "success": True,
+                    "image": {
+                        "root_index": index,
+                        "path": Path(relative).as_posix(),
+                        "filename": path.name,
+                    },
+                    "prompts": prompts,
+                }
+            )
+        except Exception:
+            self.logger.exception("Prompt inspection failed")
+            return web.json_response(
+                {"success": False, "error": "Prompt inspection failed"}, status=500
+            )
+
    async def get_image_prompt(self, request):
        """Get prompt information for a specific image path."""
        try:
@@ -1529,6 +1798,26 @@ class ImageRoutesMixin:
            if _in_subfolder(str(entry[0].relative_to(entry[2]).parent), subfolder)
        ]

+    def _format_workspace_page(self, page: list[tuple]) -> list[dict]:
+        images = self._format_output_page(page)
+        paths = {
+            (entry[3], entry[0].relative_to(entry[2]).as_posix()): entry[0]
+            for entry in page
+        }
+        for image in images:
+            image["width"] = None
+            image["height"] = None
+            if image["is_video"]:
+                continue
+            try:
+                with paths[(image["root_index"], image["path"])].open("rb") as stream:
+                    with Image.open(stream) as source:
+                        image["width"], image["height"] = source.size
+            except (OSError, ValueError, Image.DecompressionBombError):
+                # A corrupt or disappearing preview must not hide the listing.
+                continue
+        return images
+
    def _format_output_page(self, page):
        entries = []
        public_roots = {}
@@ -0,0 +1,255 @@
+"""Read-only native library with bounded, filter-scoped continuation cursors."""
+
+import base64
+import binascii
+import hashlib
+import hmac
+import json
+import os
+import time
+from collections import OrderedDict
+from collections.abc import Callable
+from pathlib import Path
+
+from aiohttp import web
+
+from .images import _canonical, _media_access_error
+from .pairs import pair_response
+
+_SECRET = os.urandom(32)
+
+
+class _NameCursorStore:
+    """Full lexical keys stay off URLs, bounded by age, count and UTF-8 bytes."""
+
+    def __init__(
+        self,
+        ttl: int = 3600,
+        max_entries: int = 128,
+        max_bytes: int = 8 * 1024 * 1024,
+        clock: Callable[[], float] = time.monotonic,
+    ) -> None:
+        self.clock = clock
+        self.ttl = ttl
+        self.max_entries = max_entries
+        self.max_bytes = max_bytes
+        self.entries = OrderedDict()
+        self.size = 0
+
+    def _remove(self, reference: str) -> None:
+        _expires, _scope, raw = self.entries.pop(reference)
+        self.size -= len(raw)
+
+    def _prune(self) -> None:
+        now = self.clock()
+        for reference, (expires, _scope, _raw) in list(self.entries.items()):
+            if expires <= now:
+                self._remove(reference)
+
+    def put(self, key: object, scope: str) -> str:
+        raw = json.dumps(key, ensure_ascii=False, allow_nan=False).encode()
+        if len(raw) > min(512 * 1024, self.max_bytes):
+            raise ValueError("Prompt name exceeds the cursor limit")
+        self._prune()
+        while self.entries and (
+            len(self.entries) >= self.max_entries
+            or self.size + len(raw) > self.max_bytes
+        ):
+            self._remove(next(iter(self.entries)))
+        reference = base64.urlsafe_b64encode(os.urandom(24)).decode()
+        self.entries[reference] = (self.clock() + self.ttl, scope, raw)
+        self.size += len(raw)
+        return reference
+
+    def get(self, reference: str, scope: str) -> object:
+        self._prune()
+        entry = self.entries.get(reference)
+        if entry is None or entry[1] != scope:
+            raise ValueError("Name cursor expired; refresh the library")
+        self.entries.move_to_end(reference)
+        return json.loads(entry[2])
+
+
+_NAME_CURSORS = _NameCursorStore()
+
+
+def _scope(value: dict) -> str:
+    return hashlib.sha256(json.dumps(value, sort_keys=True).encode()).hexdigest()
+
+
+def _encode(key: object, scope: str, *, name: bool = False) -> str | None:
+    if key is None:
+        return None
+    value = (
+        {"scope": scope, "reference": _NAME_CURSORS.put(key, scope)}
+        if name
+        else {"scope": scope, "key": key}
+    )
+    payload = json.dumps(value, separators=(",", ":"), allow_nan=False).encode()
+    signature = hmac.new(_SECRET, payload, hashlib.sha256).digest()
+    return base64.urlsafe_b64encode(signature + payload).decode().rstrip("=")
+
+
+def _decode(token: str | None, scope: str) -> object:
+    if token is None:
+        return None
+    try:
+        if not 1 <= len(token) <= 4096:
+            raise ValueError
+        decoded = base64.b64decode(
+            token + "=" * (-len(token) % 4), altchars=b"-_", validate=True
+        )
+        signature, payload = decoded[:32], decoded[32:]
+        if not hmac.compare_digest(
+            signature, hmac.new(_SECRET, payload, hashlib.sha256).digest()
+        ):
+            raise ValueError
+        value = json.loads(payload)
+        if (
+            set(value) not in ({"scope", "key"}, {"scope", "reference"})
+            or value["scope"] != scope
+        ):
+            raise ValueError
+        if "reference" in value:
+            return _NAME_CURSORS.get(value["reference"], scope)
+        return value["key"]
+    except (ValueError, TypeError, binascii.Error, UnicodeError):
+        raise ValueError("Invalid library cursor; refresh the library") from None
+
+
+def _positive_id(request: web.Request) -> int:
+    value = int(request.match_info["positive_id"])
+    if not 1 <= value <= 2**63 - 1:
+        raise ValueError("Invalid positive prompt ID")
+    return value
+
+
+def _page_limit(request: web.Request, key: str = "limit") -> int:
+    value = int(request.query.get(key, "40"))
+    if not 1 <= value <= 200:
+        raise ValueError("limit must be between 1 and 200")
+    return value
+
+
+def _missing() -> web.Response:
+    return web.json_response(
+        {"success": False, "error": "Positive prompt not found"}, status=404
+    )
+
+
+class LibraryRoutesMixin:
+    def _register_library_routes(self, routes: web.RouteTableDef) -> None:
+        routes.get("/prompt_manager/library")(self.list_library)
+        routes.get("/prompt_manager/library/options")(self.get_library_options)
+        routes.get("/prompt_manager/library/{positive_id}")(self.get_library_positive)
+        routes.get("/prompt_manager/library/{positive_id}/images")(
+            self.get_library_images
+        )
+
+    def _library_media(self, records: list[dict], roots: list[Path]) -> list[dict]:
+        page = []
+        for record in records:
+            try:
+                raw_path = Path(record["image_path"])
+                if not raw_path.is_absolute():
+                    continue
+                path = _canonical(raw_path)
+                denied = _media_access_error(path, roots)
+                if denied is not None or not path.is_file():
+                    continue
+                for index, root in enumerate(roots):
+                    canonical_root = _canonical(root)
+                    if path.is_relative_to(canonical_root):
+                        page.append((path, 0, canonical_root, index))
+                        break
+            except (OSError, ValueError, RuntimeError):
+                continue
+        return self._format_workspace_page(page)
+
+    @pair_response
+    async def get_library_options(self, request: web.Request) -> web.Response:
+        result = await self._run_in_executor(self.db.get_library_options)
+        return web.json_response({"success": True, **result})
+
+    @pair_response
+    async def list_library(self, request: web.Request) -> web.Response:
+        options = {
+            "text": request.query.get("text", ""),
+            "category": request.query.get("category"),
+            "tags": sorted(set(request.query.getall("tag", []))),
+            "sort": request.query.get("sort", "newest"),
+        }
+        scope = _scope({"kind": "parents", **options})
+        result = await self._run_in_executor(
+            self.db.list_library,
+            **options,
+            limit=_page_limit(request),
+            after=_decode(request.query.get("cursor"), scope),
+        )
+        roots = list(self._get_all_output_dirs())
+
+        def previews() -> None:
+            records = self.db.get_library_previews(
+                [parent["id"] for parent in result["items"]]
+            )
+            for parent in result["items"]:
+                parent["previews"] = self._library_media(records[parent["id"]], roots)
+
+        await self._run_in_executor(previews)
+        result["next_cursor"] = _encode(
+            result.pop("next_key"), scope, name=options["sort"] == "name"
+        )
+        return web.json_response({"success": True, **result})
+
+    @pair_response
+    async def get_library_positive(self, request: web.Request) -> web.Response:
+        positive = _positive_id(request)
+        scope = _scope({"kind": "variants", "positive": positive})
+        result = await self._run_in_executor(
+            self.db.get_library_positive,
+            positive,
+            variant_limit=_page_limit(request, "variant_limit"),
+            variant_after=_decode(request.query.get("variant_cursor"), scope),
+        )
+        if result is None:
+            return _missing()
+        roots = list(self._get_all_output_dirs())
+        records = await self._run_in_executor(
+            self.db.list_library_images, result["positive"]["id"], limit=2
+        )
+        result["positive"]["previews"] = await self._run_in_executor(
+            self._library_media, records["images"], roots
+        )
+        result["variants_next_cursor"] = _encode(result.pop("variants_next_key"), scope)
+        return web.json_response({"success": True, **result})
+
+    @pair_response
+    async def get_library_images(self, request: web.Request) -> web.Response:
+        positive = _positive_id(request)
+        pair = request.query.get("pair_id")
+        pair = int(pair) if pair is not None else None
+        if pair is not None and not 1 <= pair <= 2**63 - 1:
+            raise ValueError("Invalid variant ID")
+        roots = list(self._get_all_output_dirs())
+        scope = _scope(
+            {
+                "kind": "images",
+                "positive": positive,
+                "pair": pair,
+                "roots": [str(_canonical(root)) for root in roots],
+            }
+        )
+        if await self._run_in_executor(self.db.get_library_positive, positive) is None:
+            return _missing()
+        result = await self._run_in_executor(
+            self.db.list_library_images,
+            positive,
+            pair_id=pair,
+            limit=_page_limit(request),
+            after=_decode(request.query.get("cursor"), scope),
+        )
+        result["images"] = await self._run_in_executor(
+            self._library_media, result["images"], roots
+        )
+        result["next_cursor"] = _encode(result.pop("next_key"), scope)
+        return web.json_response({"success": True, **result})
@@ -0,0 +1,204 @@
+"""Reusable ordered prompt combinations, separate from legacy prompt CRUD."""
+
+from functools import wraps
+from typing import Callable
+
+from aiohttp import web
+
+from .prompts import parse_page_params
+
+try:
+    from ...database.pairs import PairDependencyError
+    from ...utils.validators import (
+        validate_category,
+        validate_prompt_text,
+        validate_rating,
+        validate_tags,
+    )
+except ImportError:
+    from database.pairs import PairDependencyError
+    from utils.validators import (
+        validate_category,
+        validate_prompt_text,
+        validate_rating,
+        validate_tags,
+    )
+
+
+def pair_response(handler: Callable) -> Callable:
+    @wraps(handler)
+    async def wrapped(self, request: web.Request) -> web.Response:
+        try:
+            return await handler(self, request)
+        except PairDependencyError as exc:
+            return web.json_response({"success": False, "error": str(exc)}, status=409)
+        except (ValueError, TypeError) as exc:
+            return web.json_response({"success": False, "error": str(exc)}, status=400)
+        except Exception:
+            self.logger.exception("Prompt pair operation failed")
+            return web.json_response(
+                {"success": False, "error": "Prompt pair operation failed"}, status=500
+            )
+
+    return wrapped
+
+
+def metadata_fields(data: dict) -> dict:
+    fields = {}
+    for name in ("title", "category", "tags", "rating", "notes"):
+        if name not in data:
+            continue
+        value = data[name]
+        if name == "tags":
+            if not isinstance(value, list):
+                raise ValueError("Tags must be a list of strings")
+            validate_tags(value)
+        elif name == "rating":
+            if isinstance(value, bool):
+                raise ValueError("Rating must be an integer")
+            validate_rating(value)
+        else:
+            if value is not None and not isinstance(value, str):
+                raise ValueError(f"{name} must be a string")
+            if name == "category":
+                validate_category(value)
+            elif value is not None and len(value) > 10000:
+                raise ValueError(f"{name} is too long")
+        fields[name] = value
+    return fields
+
+
+async def object_body(request: web.Request) -> dict:
+    data = await request.json()
+    if not isinstance(data, dict):
+        raise ValueError("Expected a JSON object")
+    return data
+
+
+def missing_pair() -> web.Response:
+    return web.json_response({"success": False, "error": "Pair not found"}, status=404)
+
+
+class PairRoutesMixin:
+    def _register_pair_routes(self, routes: web.RouteTableDef) -> None:
+        routes.get("/prompt_manager/pairs/export")(self.export_prompt_pairs)
+        routes.get("/prompt_manager/pairs")(self.list_prompt_pairs)
+        routes.post("/prompt_manager/pairs")(self.save_prompt_pair)
+        routes.get("/prompt_manager/pairs/{pair_id}")(self.get_prompt_pair)
+        routes.patch("/prompt_manager/pairs/{pair_id}")(self.update_prompt_pair)
+        routes.delete("/prompt_manager/pairs/{pair_id}")(self.delete_prompt_pair)
+
+    @pair_response
+    async def save_prompt_pair(self, request: web.Request) -> web.Response:
+        data = await object_body(request)
+        positive = data.get("positive_text")
+        negative = data.get("negative_text")
+        validate_prompt_text(positive)
+        if not isinstance(negative, str):
+            raise ValueError(
+                "negative_text must be explicit; use an empty string for none"
+            )
+        if negative.strip():
+            validate_prompt_text(negative)
+        fields = metadata_fields(data)
+        pair_id = await self._run_in_executor(
+            self.db.save_prompt_pair, positive, negative, **fields
+        )
+        pair = await self._run_in_executor(self.db.get_prompt_pair, pair_id)
+        return web.json_response({"success": True, "pair": pair})
+
+    @pair_response
+    async def get_prompt_pair(self, request: web.Request) -> web.Response:
+        pair = await self._run_in_executor(
+            self.db.get_prompt_pair, int(request.match_info["pair_id"])
+        )
+        return (
+            web.json_response({"success": True, "pair": pair})
+            if pair
+            else missing_pair()
+        )
+
+    @pair_response
+    async def list_prompt_pairs(self, request: web.Request) -> web.Response:
+        limit, offset = parse_page_params(request.query)
+        archived = request.query.get("include_archived", "false")
+        if archived not in ("true", "false"):
+            raise ValueError("include_archived must be true or false")
+        result = await self._run_in_executor(
+            self.db.list_prompt_pairs,
+            text=request.query.get("text"),
+            category=request.query.get("category"),
+            tags=request.query.getall("tag", []),
+            include_archived=archived == "true",
+            sort=request.query.get("sort", "newest"),
+            limit=limit,
+            offset=offset,
+        )
+        return web.json_response({"success": True, **result})
+
+    @pair_response
+    async def update_prompt_pair(self, request: web.Request) -> web.Response:
+        pair_id = int(request.match_info["pair_id"])
+        data = await object_body(request)
+        allowed = {"title", "category", "tags", "rating", "notes", "archived"}
+        if not data or set(data) - allowed:
+            raise ValueError(
+                "Update metadata or archive; save a new pair to change text"
+            )
+        fields = metadata_fields(data)
+        if "archived" in data:
+            if not isinstance(data["archived"], bool) or fields:
+                raise ValueError("Archive must be a separate boolean update")
+            updated = await self._run_in_executor(
+                self.db.archive_prompt_pair, pair_id, data["archived"]
+            )
+        else:
+            updated = await self._run_in_executor(
+                self.db.update_prompt_pair_metadata, pair_id, **fields
+            )
+        pair = await self._run_in_executor(self.db.get_prompt_pair, pair_id)
+        if not updated and not pair:
+            return missing_pair()
+        return web.json_response({"success": True, "pair": pair})
+
+    @pair_response
+    async def delete_prompt_pair(self, request: web.Request) -> web.Response:
+        deleted = await self._run_in_executor(
+            self.db.delete_prompt_pair, int(request.match_info["pair_id"])
+        )
+        return web.json_response({"success": True}) if deleted else missing_pair()
+
+    @pair_response
+    async def export_prompt_pairs(self, request: web.Request) -> web.Response:
+        pairs = []
+        offset = 0
+        while True:
+            result = await self._run_in_executor(
+                self.db.list_prompt_pairs,
+                include_archived=True,
+                limit=500,
+                offset=offset,
+            )
+            for pair in result["pairs"]:
+                pairs.append(
+                    {
+                        key: pair[key]
+                        for key in (
+                            "positive_text",
+                            "negative_text",
+                            "title",
+                            "category",
+                            "tags",
+                            "rating",
+                            "notes",
+                            "archived",
+                        )
+                    }
+                )
+            offset += len(result["pairs"])
+            if not result["has_more"] or not result["pairs"]:
+                break
+        return web.json_response(
+            {"schema_version": 1, "kind": "prompt-manager-pairs", "pairs": pairs},
+            headers={"Content-Disposition": 'attachment; filename="prompt-pairs.json"'},
+        )
@@ -6,6 +6,11 @@ import os
 import sqlite3

 from aiohttp import web
+
+try:
+    from ...database.pairs import PairDependencyError
+except ImportError:
+    from database.pairs import PairDependencyError
 from urllib.parse import quote

 # Upper bounds on page sizes and offsets for every list endpoint. The offset
@@ -793,6 +798,8 @@ class PromptRoutesMixin:
                    status=404,
                )

+        except PairDependencyError as exc:
+            return web.json_response({"success": False, "error": str(exc)}, status=409)
        except ValueError:
            return web.json_response(
                {"success": False, "error": "Invalid prompt ID"}, status=400
@@ -840,6 +847,8 @@ class PromptRoutesMixin:
                    {"success": False, "error": "Prompt not found"}, status=404
                )

+        except PairDependencyError as exc:
+            return web.json_response({"success": False, "error": str(exc)}, status=409)
        except sqlite3.IntegrityError:
            return web.json_response(
                {"success": False, "error": "Another prompt already has this text"},
@@ -1081,6 +1090,8 @@ class PromptRoutesMixin:
                }
            )

+        except PairDependencyError as exc:
+            return web.json_response({"success": False, "error": str(exc)}, status=409)
        except Exception as e:
            self.logger.error(f"Bulk delete error: {e}")
            return web.json_response(
@@ -16,6 +16,7 @@ class ScanJob:
    def __init__(self):
        self.last_event = None
        self.finished = False
+        self.cancelled = False
        self.task = None
        self._subscribers = set()

@@ -0,0 +1 @@
+"""Browser workflow fixtures; never imported by production code."""
--- a/Show More
+++ b/Show More
Author	SHA1	Message	Date
Vito Sansevero	03a8ccc10e	fix(history): honor close deadlines after early joins (#209 )	2026-10-17 19:38:29 -07:00
Vito Sansevero	70a418ecac	fix(history): harden context validation and worker fixtures (#209 )	2026-10-17 19:19:53 -07:00
Vito Sansevero	8890c56c30	test(ci): update execution context tooling fixtures (#209 )	2026-10-17 19:11:04 -07:00
Vito Sansevero	2fcbf5389d	fix(history): scope transforms to captured source paths	2026-10-17 19:03:56 -07:00
Vito Sansevero	20c8e9eef9	test(history): verify durable contexts on real ComfyUI hosts	2026-10-17 18:54:53 -07:00
Vito Sansevero	0bdb33230f	feat(history): capture cached prompt strings without mutating host caches	2026-10-17 18:38:56 -07:00
Vito Sansevero	436ae95cda	test(history): discover owner invalidation callback regression	2026-10-17 18:30:55 -07:00
Vito Sansevero	da4d6cdc84	feat(history): bind exact runtime context to execution attempts	2026-10-17 18:28:23 -07:00
Vito Sansevero	8e014c0ebe	fix(history): bound persisted identifier bytes before restore reads	2026-10-17 18:10:37 -07:00
Vito Sansevero	8b922c6b64	fix(history): fence context commits and resolve stored run identity	2026-10-17 18:04:48 -07:00
Vito Sansevero	98a8475d10	feat(history): bound context payloads beside lifecycle finalization	2026-10-17 17:58:38 -07:00
Vito Sansevero	99e4d09419	fix(history): preserve package identity for execution contexts	2026-10-17 17:51:35 -07:00
Vito Sansevero	600a479f0b	feat(history): persist immutable execution context snapshots	2026-10-17 17:47:27 -07:00
Vito Sansevero	cc6e7f9ec0	feat(history): migrate execution contexts to schema v2	2026-10-17 17:39:20 -07:00
Vito Sansevero	523c136455	fix(history): retain observable sources and validate evidence	2026-10-17 17:29:27 -07:00
Vito Sansevero	08e77de05f	feat(history): define exact bounded execution contexts	2026-10-17 17:24:16 -07:00
Vito	763e6b37b1	feat(history): record durable ComfyUI execution attempts (#212 ) * feat(history): normalize bounded execution evidence (#209) * fix(history): bound lifecycle kind checks before hashing (#209) * feat(history): persist execution attempts with restore fencing (#209) * test(api): bootstrap standalone backup restore fixtures (#209) * feat(history): isolate bounded asynchronous ledger writes (#209) * feat(history): observe actual queue attempts without altering execution (#209) * fix(database): honor package-scoped configuration (#211) * test(history): exercise real cached and headless executions (#209) * fix(history): reject unsupported uniqueness and isolate package fixtures (#209) * test(history): support Python 3.10 package fixture cleanup (#209)	2026-10-17 16:42:07 -07:00
Vito	fdba6fd012	feat(workspace): group saved negatives in the native Prompt Library (#210 ) * feat(workspace): group saved negative variants in native library (#208) * fix(library): preserve full name order and batch previews (#208)	2026-10-17 14:38:03 -07:00
Vito	dc345671ae	feat(workspace): inspect and send positive and negative prompts (#206 ) * feat(workspace): inspect and send positive and negative prompts * docs(workspace): capture complete inspector and viewer previews * docs(inspector): explain the subprocess cleanup race * fix(workspace): defer observed gallery layout writes (#207) * fix(inspector): keep path-only prompt associations unverified	2026-10-17 13:43:19 -07:00
Vito	76e9b51079	feat(bridge): add acknowledged ComfyUI prompt destinations (#203 ) * feat(bridge): add guarded PromptManager prompt destinations * fix(ci): install shared contracts before frontend unit tests * test(comfyui): wait for paired editor resize before measuring layout	2026-10-17 13:14:02 -07:00
Vito	590aa2cd0e	feat(workspace): add native Images browsing and infinite scroll (#201 ) * feat(workspace): add native image browsing and continuous loading * test(workspace): assert empty collections through their status * fix(workspace): rescan refreshed images and expose ancestor folders	2026-10-17 11:22:10 -07:00
Vito	d4bb5c5ba9	feat(ui): add the tested opt-in Prompt Manager workspace shell (#198 ) * docs(ui): prototype unified prompt management application * docs(ui): add masonry and per-image tagging to application concept * docs(ui): make gallery concept use continuous infinite scrolling * feat(ui): introduce the tested application workspace shell * fix(docs): preserve concept controls across cached navigation * test(ci): emit Python tracebacks for native test crashes * test(ui): cover workspace restoration across cached navigation	2026-10-17 10:36:30 -07:00
Vito	caa9649f32	feat: reuse positive and negative prompt combinations (#195 ) * feat(db): add pair schema and stage restore migrations for #193 * feat(database): persist reusable ordered prompt pairs (#193) * feat(api): expose reusable pairs and atomic metadata enrichment (#193) * feat(metadata): resolve canonical prompt pairs with bounded provenance (#193) * feat(scan): enrich image pairs without guessing or duplicate links (#193) * fix(database): preserve pair tags during global maintenance (#193) * feat(tracking): enrich live image pairs and cancel scans safely (#193) * feat(nodes): encode paired prompts and count cached recipe runs (#193) * fix(database): validate restored pair ownership and provenance (#193) * feat(admin): report pair scan outcomes and cancellation (#193) * feat(ui): add paired prompt editor and explicit recipe library for #193 * fix(autocomplete): include reusable pair metadata vocabulary (#193) * fix(pairs): honor recent sorting and conservative scan outcomes (#193) * fix(ui): protect paired draft saves and clarify negative variants (#193) * fix(metadata): preserve bounded paired execution evidence (#193) * fix(tracking): follow actual pair output roles and prefer execution evidence (#193) * docs(pairs): document live validation and retain renderer screenshots (#193) * test(api): inject disposable databases before API startup (#194) * test(api): canonicalize portable temporary gallery roots (#194) * docs(ui): prototype branching negative variant history (#193) * docs(ui): show selected negative beneath positive in timeline concept (#193) * docs: retain approved timeline for later gallery redesign (#193) * test(db): close restore fixture connections before cleanup (#194) * test(db): reuse existing handle guards in all pair fixtures (#194) * chore: clarify rollback test and intentional exception handling (#193)	2026-10-17 08:00:42 -07:00
Vito	3870590704	feat(ui): add Split Prompt Manager workspace (#192 ) * feat(ui): add Split Prompt Manager workspace (#186) * test(ui): use consistent unittest imports (#186)	2026-10-17 06:47:46 -07:00
Vito	a52ff4e052	feat(ui): add Focused Prompt Manager with shared autocomplete (#191 ) * feat(ui): add Focused Prompt Manager node (#186) * feat(ui): autocomplete saved Focused metadata with shared controls (#186) * feat(ui): autocomplete Focused library filters (#186)	2026-10-17 05:31:24 -07:00
Vito	035e816db6	feat(ui): adopt shared toolkit for Prompt Manager Classic (#190 ) * feat(ui): adopt shared toolkit for Prompt Manager Classic (#186) * test(ui): retain Text node Web UI coverage alongside Classic * fix(tooling): reject lookalike toolkit download URLs (#186)	2026-10-16 22:18:18 -07:00
Vito	f98f6fa08d	test(ci): validate shared toolkit in real PromptManager hosts (#189 ) * test(ci): prove shared toolkit with real PromptManager host coverage * test(ci): match KikoTools Python 3.13 compatibility coverage	2026-10-16 20:56:50 -07:00
Vito	ce82dfb912	fix: quiet routine console logging and prevent duplicate output (#183 )	2026-10-16 09:58:05 -07:00
Vito	5ded3189c1	test: cover backup restore and destructive browser workflows (#185 )	2026-10-16 07:56:15 -07:00
Vito	f1b1d4731b	test: exercise real page workflows and enforce frontend coverage (#181 )	2026-10-16 06:45:20 -07:00
				`@@ -0,0 +1 @@`
				`"""Browser workflow fixtures; never imported by production code."""`