fix(history): join case-variant raw text to the history of the prompt it counts toward (#460) (#471)

* fix(history): join outer-whitespace raw text to the history of the prompt it saves (#438)

The Library stores text.strip() while a run keeps its exact raw text. The
history window, run detail scope id, newest pair run and prompt hints now
compare the raw text's stored form (one rule, utils/hashing.py, mirrored in
SQLite with the same code points). Hints also keep matching the exact raw
text, so older rows that kept outer whitespace stay hinted and protected.

* fix(history): join case-variant raw text to the history of the prompt it counts toward (#460)

The prompt hash folds case, so a run of "A Lighthouse" counts toward the
saved "a lighthouse" but never joined its history. History now compares the
hash's own key (prompt_key = strip + str.lower()); SQL calls Python for it
through pm_prompt_key/pm_prompt_hash registered on every managed connection,
since SQLite's lower() folds ASCII only. Hints match by text first (#438) and
fall back to the dedup hash only when no text matches, so no existing hint
becomes ambiguous. A history position from before this change still resumes.

* test(history): a real-host case-variant re-queue joins the prompt's history (#460)

database/execution_contexts.py

@@ -7,7 +7,11 @@ from typing import Callable, ContextManager, Iterator
 
 try:
     from ..utils.execution_history_events import RunFinish, encode_command
-    from ..utils.hashing import stored_prompt_text, stored_prompt_text_sql
+    from ..utils.hashing import (
+        generate_prompt_hash,
+        stored_prompt_text,
+        stored_prompt_text_sql,
+    )
     from ..utils.execution_contexts import (
         CAPTURE_REASONS,
         ContextBundle,
@@ -16,7 +20,11 @@ try:
     )
 except ImportError:
     from utils.execution_history_events import RunFinish, encode_command
-    from utils.hashing import stored_prompt_text, stored_prompt_text_sql
+    from utils.hashing import (
+        generate_prompt_hash,
+        stored_prompt_text,
+        stored_prompt_text_sql,
+    )
     from utils.execution_contexts import (
         CAPTURE_REASONS,
         ContextBundle,
@@ -149,9 +157,12 @@ def _role_hints(role: str) -> str:
 
 # Set form of the read-time prompt-hint rule in get_execution_contexts():
 # a stored role hint is live only in the current storage epoch, and only when
-# exactly one prompt has its raw snapshot text or that text's stored form
-# (outer whitespace stripped, #438), and that is the hinted one (_prompt_hint).
-# Evaluated once per statement, never per prompt row.
+# the hinted prompt is the one match of _prompt_hint: by text (the raw
+# snapshot text or its stored form, outer whitespace stripped, #438), or, only
+# when no prompt has either text, by dedup hash (case folded too, #460).
+# The hash is a fallback, so a hint #438 kept is never made ambiguous.
+# Needs register_prompt_functions (pm_prompt_hash). The hint rows are the
+# outer loop (CROSS JOIN), so each prompts lookup is by index.
 # Defines the CTE ``live_hint_prompts(prompt_id)``; prefix it with WITH.
 LIVE_HINT_CTES = (
     "stored_hints(prompt_id, raw, stored) AS ("
@@ -160,7 +171,12 @@ LIVE_HINT_CTES = (
     + _role_hints("negative")
     + "), hint_matches(prompt_id, raw, stored, match_id) AS ("
     "SELECT h.prompt_id, h.raw, h.stored, p.id FROM stored_hints h "
-    "JOIN prompts p ON p.text COLLATE BINARY IN (h.raw, h.stored)"
+    "CROSS JOIN prompts p ON p.text COLLATE BINARY IN (h.raw, h.stored) "
+    "UNION ALL "
+    "SELECT h.prompt_id, h.raw, h.stored, p.id FROM stored_hints h "
+    "CROSS JOIN prompts p ON p.hash = pm_prompt_hash(h.raw) "
+    "WHERE NOT EXISTS (SELECT 1 FROM prompts q "
+    "WHERE q.text COLLATE BINARY IN (h.raw, h.stored))"
     "), live_hint_prompts(prompt_id) AS ("
     "SELECT prompt_id FROM hint_matches GROUP BY prompt_id, raw, stored "
     "HAVING COUNT(*) = 1 AND MIN(match_id) = prompt_id)"
@@ -169,20 +185,32 @@ LIVE_HINT_PROMPT_IDS_SQL = (
     "WITH " + LIVE_HINT_CTES + " SELECT DISTINCT prompt_id FROM live_hint_prompts"
 )
 
+_BY_TEXT = "SELECT id FROM prompts WHERE text COLLATE BINARY IN (?,?) LIMIT 2"
+
+
+def _text_matches(conn: sqlite3.Connection, raw: str) -> list:
+    return [row[0] for row in conn.execute(_BY_TEXT, (raw, stored_prompt_text(raw)))]
+
 
 def _prompt_hint(conn: sqlite3.Connection, raw: str | None) -> int | None:
-    """The one prompt whose text is ``raw`` or the stored form of it (#438).
+    """The one prompt that stores ``raw`` (#438), else the one it counts toward.
 
-    The Library stores ``text.strip()``; the exact form keeps hints of older
-    rows whose stored text kept outer whitespace.
+    By text first: ``raw`` or ``raw.strip()`` (the exact form keeps hints of
+    older rows that kept outer whitespace). Only when no prompt has either
+    text, by the dedup hash, which also folds case (#460).
     """
     if raw is None:
         return None
-    rows = conn.execute(
-        "SELECT id FROM prompts WHERE text COLLATE BINARY IN (?,?) LIMIT 2",
-        (raw, stored_prompt_text(raw)),
-    ).fetchall()
-    return rows[0][0] if len(rows) == 1 else None
+    matches = _text_matches(conn, raw)
+    if not matches:
+        matches = [
+            row[0]
+            for row in conn.execute(
+                "SELECT id FROM prompts WHERE hash=? LIMIT 2",
+                (generate_prompt_hash(raw),),
+            )
+        ]
+    return matches[0] if len(matches) == 1 else None
 
 
 def _hints(
@@ -191,24 +219,16 @@ def _hints(
     pos_id, neg_id = _prompt_hint(conn, positive), _prompt_hint(conn, negative)
     stored = None if negative is None else stored_prompt_text(negative)
     pair_id = None
+    # A known negative that names no prompt because several share its text
     negative_ambiguous = (
-        stored is not None
-        and neg_id is None
-        and conn.execute(
-            "SELECT 1 FROM prompts WHERE text COLLATE BINARY IN (?,?) LIMIT 1",
-            (negative, stored),
-        ).fetchone()
-        is not None
+        stored is not None and neg_id is None and bool(_text_matches(conn, negative))
     )
     if pos_id is not None and stored 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 "
+            "SELECT pp.id FROM prompt_pairs pp "
             "WHERE pp.positive_prompt_id=? AND ((pp.negative_prompt_id IS NULL AND "
-            "?='') "
-            "OR (pp.negative_prompt_id=? AND n.text COLLATE BINARY IN (?,?))) "
-            "LIMIT 2",
-            (pos_id, stored, neg_id, negative, stored),
+            "?='') OR pp.negative_prompt_id=?) LIMIT 2",
+            (pos_id, stored, neg_id),
         ).fetchall()
         if len(rows) == 1:
             pair_id = rows[0][0]

database/health.py

@@ -23,10 +23,10 @@ from pathlib import PurePosixPath, PureWindowsPath
 from urllib.parse import quote
 
 try:
-    from ..utils.hashing import generate_prompt_hash
+    from ..utils.hashing import register_prompt_functions
     from .models import normalize_image_path
 except ImportError:
-    from utils.hashing import generate_prompt_hash
+    from utils.hashing import register_prompt_functions
     from database.models import normalize_image_path
 
 logger = logging.getLogger("prompt_manager.database.health")
@@ -125,12 +125,8 @@ def _foreign_keys(conn, _roots):
     )
 
 
-def _prompt_hash(text):
-    return generate_prompt_hash(text) if isinstance(text, str) else None
-
-
 def _canonical_hashes(conn, _roots):
-    conn.create_function("pm_prompt_hash", 1, _prompt_hash, deterministic=True)
+    register_prompt_functions(conn)  # pm_prompt_hash, shared with history (#460)
     return _counted(
         f"Prompt {row[0]}: the stored hash does not match its text"
         for row in conn.execute(

database/history_reads.py

@@ -16,9 +16,9 @@ from .execution_outputs import _capture, _REFERENCE_COLUMNS
 from .library import _ELIGIBLE
 
 try:
-    from ..utils.hashing import stored_prompt_text, stored_prompt_text_sql
+    from ..utils.hashing import prompt_key, stored_prompt_text
 except ImportError:
-    from utils.hashing import stored_prompt_text, stored_prompt_text_sql
+    from utils.hashing import prompt_key, stored_prompt_text
 
 RunKey = tuple[int, str, int, str]
 
@@ -115,12 +115,12 @@ _MATCHING = (
 )
 
 
-def _scope_text_ids(conn: sqlite3.Connection, scope: str) -> str:
-    """Every recorded raw text whose stored form is ``scope``, as a JSON array.
+def _scope_text_ids(conn: sqlite3.Connection, key: str) -> str:
+    """Every recorded raw text whose prompt key is ``key``, as a JSON array.
 
-    The Library stores ``text.strip()`` while a run keeps its exact raw text,
-    so a raw text joins the scope of the prompt it saves (#438). The values
-    are compared, never only their digests.
+    A run keeps its exact raw text, while the Library saves and counts it as
+    the prompt with its key (outer whitespace #438, case #460). The values are
+    compared, never only their digests; ``pm_prompt_key`` is Python's key.
     """
     return json.dumps(
         [
@@ -130,16 +130,16 @@ def _scope_text_ids(conn: sqlite3.Connection, scope: str) -> str:
                 "SELECT t.text_id FROM (SELECT DISTINCT raw_text_id AS text_id "
                 "FROM execution_prompt_snapshots) s CROSS JOIN "
                 "execution_context_texts t WHERE t.text_id=s.text_id AND "
-                f"{stored_prompt_text_sql('t.text_value')}=? COLLATE BINARY",
-                (scope,),
+                "pm_prompt_key(t.text_value)=? COLLATE BINARY",
+                (key,),
             )
         ]
     )
 
 
 def _scope_text_id(raw: str | None) -> str | None:
-    """The scope digest of a raw text: that of the text the Library stores."""
-    return None if raw is None else content_digest("text", (stored_prompt_text(raw),))
+    """The scope digest of a raw text: that of its prompt key."""
+    return None if raw is None else content_digest("text", (prompt_key(raw),))
 
 
 def _limit(limit: int) -> None:
@@ -290,7 +290,14 @@ class HistoryReadOperations:
             return None
         text = saved["text"]
         raw_id = _scope_text_id(text)
-        if after is not None and after.raw_text_id != raw_id:
+        # A position from before #460 names the scope by the stripped text
+        # alone; it is the same scope, so a page open across the upgrade resumes.
+        legacy_id = (
+            None
+            if text is None
+            else content_digest("text", (stored_prompt_text(text),))
+        )
+        if after is not None and after.raw_text_id not in (raw_id, legacy_id):
             raise ValueError("History position text scope changed")
         group = {
             "positive_id": positive_id,
@@ -309,7 +316,7 @@ class HistoryReadOperations:
         )
         if raw_id is None:
             return result
-        scope = _scope_text_ids(conn, stored_prompt_text(text))
+        scope = _scope_text_ids(conn, prompt_key(text))
         where = f"r.run_id IN ({_MATCHING})"
         params = [scope]
         if after is not None:
@@ -465,7 +472,7 @@ class HistoryReadOperations:
         if saved["text"] is None:
             return {"run": None}
         epoch = self.get_execution_epoch()
-        stored = stored_prompt_text(saved["text"])
+        key = prompt_key(saved["text"])
         candidates = conn.execute(
             "SELECT r.run_id,rc.anchor_id,rc.context_id,"
             "c.positive_snapshot_id,c.negative_snapshot_id "
@@ -477,7 +484,7 @@ class HistoryReadOperations:
             "AND rc.reference_epoch=? "
             f"ORDER BY {_ORDER},rc.anchor_id COLLATE BINARY LIMIT ?",
             (
-                _scope_text_ids(conn, stored),
+                _scope_text_ids(conn, key),
                 pair_id,
                 epoch,
                 _PAIR_RUN_CANDIDATES,
@@ -489,7 +496,7 @@ class HistoryReadOperations:
             if (
                 positive is None
                 or positive["raw"] is None
-                or stored_prompt_text(positive["raw"]) != stored
+                or prompt_key(positive["raw"]) != key
             ):
                 continue
             hints = _hints(conn, positive["raw"], negative["raw"] if negative else None)

database/maintenance.py

@@ -44,6 +44,11 @@ from .health import BATCH_ROWS, link_states, readable_roots, readonly_uri
 from .pairs import PairDependencyError
 from .workflow_documents import documents_of_rows, release_documents
 
+try:
+    from ..utils.hashing import register_prompt_functions
+except ImportError:
+    from utils.hashing import register_prompt_functions
+
 MAINTENANCE_OPERATIONS = (
     "duplicate_prompts",
     "optimize",
@@ -289,6 +294,7 @@ def _read_snapshot(db_path, read):
     conn = sqlite3.connect(readonly_uri(db_path), uri=True)
     try:
         conn.row_factory = sqlite3.Row
+        register_prompt_functions(conn)  # LIVE_HINT_CTES calls them (#460)
         conn.execute("BEGIN")  # one read snapshot for the whole plan
         return read(conn)
     finally:

database/models.py

@@ -38,6 +38,11 @@ from .review_schema import (
     review_without_annotations_error,
 )
 from .saved_views import migrate_saved_views, saved_views_schema_error
+
+try:
+    from ..utils.hashing import register_prompt_functions
+except ImportError:
+    from utils.hashing import register_prompt_functions
 from .image_categories import image_categories_schema_error, migrate_image_categories
 from .workflow_documents import (
     DOCUMENT_ID_COLUMNS,
@@ -153,12 +158,15 @@ def configure_connection(conn: sqlite3.Connection) -> None:
     (common on Windows) let other writers' polling busy handlers starve past
     BUSY_TIMEOUT_SECONDS and fail with "database is locked" (#242). Without
     WAL (for example an in-memory database) SQLite's FULL default is kept.
+    It also registers the prompt-key SQL functions (register_prompt_functions).
     """
     journal_mode = conn.execute("PRAGMA journal_mode = WAL").fetchone()[0]
     conn.execute("PRAGMA foreign_keys = ON")
     conn.execute(f"PRAGMA busy_timeout = {BUSY_TIMEOUT_SECONDS * 1000}")
     if str(journal_mode).lower() == "wal":
         conn.execute("PRAGMA synchronous = NORMAL")
+    # History reads and the live-hint rule call Python's prompt key (#460)
+    register_prompt_functions(conn)
 
 
 class PromptModel:

tests/comfyui/execution-history.spec.mjs

@@ -314,6 +314,16 @@ test('fresh registered PM variants preserve exact raw whitespace and known empty
         const runId = (await readExecutionRuns(execution.databasePath, id)).rows[0].run_id;
         const response = await execution.request.get(`/prompt_manager/library/${value.positive_prompt_id}/history`);expect(response.ok()).toBe(true);
         expect((await response.json()).items.map(item => item.run_id)).toContain(runId);
+        if (kind !== 'PromptManager') continue;
+        // #460: an upper-case re-queue counts toward the same prompt and joins its history
+        const upper = await submit(execution.request, recipeGraph({ kind, text: `${kind.toUpperCase()} E\u0301` }));
+        await history(execution.request, upper);
+        const variant = (await capture(execution, upper)).contexts[0];
+        expect(variant.positive.raw).toBe(`${kind.toUpperCase()} E\u0301`);
+        expect(variant.positive_prompt_id, 'the case variant names the same prompt').toBe(value.positive_prompt_id);
+        const variantRun = (await readExecutionRuns(execution.databasePath, upper)).rows[0].run_id;
+        const scoped = await execution.request.get(`/prompt_manager/library/${value.positive_prompt_id}/history`);
+        expect((await scoped.json()).items.map(item => item.run_id)).toContain(variantRun);
     }
 });
 

tests/test_api_history.py

@@ -18,6 +18,8 @@ from unittest.mock import patch
 
 from aiohttp import FormData, web
 
+from utils.hashing import generate_prompt_hash
+
 try:
     from test_api_images import ImageAPITestCase
 except ImportError:
@@ -191,7 +193,11 @@ class HistoryAPITests(ImageAPITestCase):
             context["evidence"]["positive"]["raw"]["provenance"], "recorded"
         )
         with self.api.db.model.get_connection() as conn:
-            conn.execute("UPDATE prompts SET text='edited' WHERE id=?", (positive,))
+            # An edit changes the text and its dedup hash together (#460)
+            conn.execute(
+                "UPDATE prompts SET text='edited',hash=? WHERE id=?",
+                (generate_prompt_hash("edited"), positive),
+            )
         _, changed = await self.detail(start.run_id)
         self.assertEqual(changed["contexts"][0]["positive"], context["positive"])
         self.assertFalse(changed["contexts"][0]["navigation"]["positive_available"])

tests/test_execution_context_database.py

@@ -15,6 +15,7 @@ from database import execution_contexts
 from database.execution_history import HistoryEpochChanged
 from database.execution_context_schema import context_schema_error
 from database.execution_contexts import LIVE_HINT_PROMPT_IDS_SQL
+from utils.hashing import generate_prompt_hash
 from database.execution_history_schema import rotate_history_epoch
 from utils.execution_contexts import (
     ContextBundle,
@@ -466,7 +467,11 @@ class ContextDatabaseTests(unittest.TestCase):
         with conn:
             conn.execute("DELETE FROM prompts WHERE id=3")
         self.assertEqual(assert_same(), {1, 2, 4})
+        # A case-only edit keeps the prompt's hash, so the run still counts
+        # toward it (#460); another text drops the hint.
         db.update_prompt_text(4, "Other")
+        self.assertEqual(assert_same(), {1, 2, 4})
+        db.update_prompt_text(4, "another")
         self.assertEqual(assert_same(), {1, 2})
         with conn:
             conn.execute("INSERT INTO prompts(id,text) VALUES(5,'none')")
@@ -522,6 +527,83 @@ class ContextDatabaseTests(unittest.TestCase):
         live = {row[0] for row in conn.execute(LIVE_HINT_PROMPT_IDS_SQL)}
         self.assertEqual(live, {1, 2})
 
+    def test_hints_follow_the_hash_for_case_variant_raw_text(self):
+        # #460: the run names the prompt the Library counted it toward.
+        db = self.database()
+        first = db.save_prompt(
+            "\u00e9t\u00e9 positive",
+            prompt_hash=generate_prompt_hash("\u00e9t\u00e9 positive"),
+        )
+        second = db.save_prompt(
+            "negative \u03c3", prompt_hash=generate_prompt_hash("negative \u03c3")
+        )
+        with db.model.get_connection() as conn:
+            conn.execute(
+                "INSERT INTO prompt_pairs(id,positive_prompt_id,negative_prompt_id) "
+                "VALUES(7,?,?)",
+                (first, second),
+            )
+        start = self.capture(
+            db, 0, bundle("NEGATIVE \u03a3", positive="\u00c9T\u00c9 Positive")
+        )
+        context = db.get_execution_contexts(start.run_id)["contexts"][0]
+        self.assertEqual(
+            (
+                context["positive_prompt_id"],
+                context["negative_prompt_id"],
+                context["pair_id"],
+            ),
+            (first, second, 7),
+        )
+        conn = db.model.get_connection()
+        live = {row[0] for row in conn.execute(LIVE_HINT_PROMPT_IDS_SQL)}
+        self.assertEqual(live, {first, second})
+        # A prompt that only shares SQLite's ASCII lower() is not a match
+        other = db.save_prompt("other", prompt_hash=generate_prompt_hash("other"))
+        start = self.capture(db, 1, bundle("x", positive="\u00d6THER"))
+        context = db.get_execution_contexts(start.run_id)["contexts"][0]
+        self.assertIsNone(context["positive_prompt_id"])
+        self.assertNotIn(
+            other, {row[0] for row in conn.execute(LIVE_HINT_PROMPT_IDS_SQL)}
+        )
+
+    def test_hash_is_only_a_fallback_so_a_text_hint_stays_unambiguous(self):
+        # #460 review: a hash match must never make a #438 text hint ambiguous,
+        # or its prompt would lose the recorded-run prune protection.
+        db = self.database()
+        conn = db.model.get_connection()
+        with conn:
+            # A legacy row without a hash, and a case variant with the hash
+            conn.execute("INSERT INTO prompts(id,text) VALUES(1,'a lighthouse')")
+            conn.execute(
+                "INSERT INTO prompts(id,text,hash) VALUES(2,'A Lighthouse',?)",
+                (generate_prompt_hash("a lighthouse"),),
+            )
+        exact = self.capture(db, 0, bundle("x", positive="a lighthouse"))
+        variant = self.capture(db, 1, bundle("x", positive="A LIGHTHOUSE"))
+        hints = [
+            db.get_execution_contexts(start.run_id)["contexts"][0]["positive_prompt_id"]
+            for start in (exact, variant)
+        ]
+        self.assertEqual(hints, [1, 2])
+        live = {row[0] for row in conn.execute(LIVE_HINT_PROMPT_IDS_SQL)}
+        self.assertEqual(live, {1, 2})
+        self.assertEqual(db.prune_orphaned_prompts(), 0)
+
+    def test_live_hint_prompt_lookups_use_indexes(self):
+        db = self.database()
+        conn = db.model.get_connection()
+        plan = [
+            row[3]
+            for row in conn.execute("EXPLAIN QUERY PLAN " + LIVE_HINT_PROMPT_IDS_SQL)
+        ]
+        prompt_steps = [
+            step for step in plan if " p " in f" {step} " or "TABLE p" in step
+        ]
+        self.assertTrue(prompt_steps, plan)
+        self.assertEqual([step for step in plan if step.startswith("SCAN p")], [])
+        self.assertEqual([step for step in plan if step.startswith("SCAN q")], [])
+
     def test_hints_require_exact_unique_raw_text_and_never_refresh(self):
         db = self.database()
         conn = db.model.get_connection()

tests/test_execution_history_hooks.py

@@ -19,6 +19,7 @@ from unittest.mock import patch
 from tests.history_waits import drain_writer
 from database import PromptDatabase
 from utils import usage_tracking
+from utils.hashing import generate_prompt_hash
 from utils.execution_history_writer import HistoryWriter
 
 try:
@@ -918,6 +919,30 @@ class CachedRuntimeHistoryTests(HooksCase):
         self.assertIsNone(named_prompt(self.db, self.rows()[1]["run_id"]))
 
 
+class CachedRuntimeCaseHistoryTests(CachedRuntimeHistoryTests):
+    """Raw text that differs only in case joins the saved prompt's history (#460).
+
+    The hash folds case, so the run counts toward the lower-case prompt saved
+    first; the run keeps its exact raw spelling.
+    """
+
+    TEXT = "\u00c9 LIGHTHOUSE the LLM wrote"
+
+    def setUp(self):
+        super().setUp()
+        self.db.delete_prompt(self.prompt_id)
+        lower = self.TEXT.lower()
+        self.prompt_id = self.db.save_prompt(
+            text=lower, prompt_hash=generate_prompt_hash(lower)
+        )
+
+    def test_the_library_counts_the_case_variant_as_this_prompt(self):
+        self.assertEqual(
+            self.db.get_prompt_by_hash(generate_prompt_hash(self.TEXT))["id"],
+            self.prompt_id,
+        )
+
+
 class CachedRuntimeWhitespaceHistoryTests(CachedRuntimeHistoryTests):
     """Raw text with outer whitespace joins the history of the prompt it saves (#438).
 

tests/test_hashing.py

@@ -17,6 +17,8 @@ from utils.hashing import (
     generate_prompt_hash,
     is_duplicate_prompt,
     STRIPPED_CODE_POINTS,
+    prompt_key,
+    register_prompt_functions,
     stored_prompt_text,
     stored_prompt_text_sql,
 )
@@ -158,5 +160,56 @@ class TestStoredPromptText(unittest.TestCase):
                 self.assertEqual(self.sql_strip(value), stored_prompt_text(value))
 
 
+class TestPromptKey(unittest.TestCase):
+    """History compares the hash's own key; SQL calls Python for it (#460)."""
+
+    CASES = (
+        "A Lighthouse",
+        " \u00c9t\u00c9\n",
+        "\u03a3\u03a3",
+        "\u0130stanbul",
+        "\u1e9e",
+        "\u212a",
+        "ok",
+    )
+
+    def test_key_is_the_hash_normalization(self):
+        import hashlib
+
+        for value in self.CASES:
+            with self.subTest(value=repr(value)):
+                self.assertEqual(
+                    generate_prompt_hash(value),
+                    hashlib.sha256(prompt_key(value).encode("utf-8")).hexdigest(),
+                )
+        self.assertEqual(prompt_key("  A Lighthouse\n"), "a lighthouse")
+
+    def test_sql_functions_give_pythons_result_where_sql_lower_does_not(self):
+        import sqlite3
+
+        conn = sqlite3.connect(":memory:")
+        try:
+            register_prompt_functions(conn)
+            for value in self.CASES:
+                with self.subTest(value=repr(value)):
+                    key, digest = conn.execute(
+                        "SELECT pm_prompt_key(?), pm_prompt_hash(?)", (value, value)
+                    ).fetchone()
+                    self.assertEqual(key, prompt_key(value))
+                    self.assertEqual(digest, generate_prompt_hash(value))
+            self.assertEqual(
+                conn.execute(
+                    "SELECT pm_prompt_key(1), pm_prompt_hash(NULL)"
+                ).fetchone(),
+                (None, None),
+            )
+            # SQLite's own lower() folds ASCII only
+            self.assertNotEqual(
+                conn.execute("SELECT lower(?)", ("\u00c9",)).fetchone()[0], "\u00e9"
+            )
+        finally:
+            conn.close()
+
+
 if __name__ == "__main__":
     unittest.main()

tests/test_history_reads.py

@@ -17,6 +17,7 @@ from utils.execution_contexts import (
 )
 from utils.execution_outputs import OutputBundle
 from tests.test_execution_context_database import bundle, snapshot
+from utils.hashing import generate_prompt_hash
 from tests.test_execution_output_database import bundle as outputs
 from tests import test_execution_history_database as history_tests
 
@@ -33,6 +34,10 @@ class HistoryReadTests(unittest.TestCase):
             conn.execute("INSERT INTO prompts(id,text) VALUES(?,?)", (identity, text))
         return identity
 
+    def hashed(self, db, text):
+        """A prompt saved as the nodes save it: with its dedup hash."""
+        return db.save_prompt(text, prompt_hash=generate_prompt_hash(text))
+
     def capture(self, db, item=0, context=None, output=None, time=100):
         start = self.start(db, item, time)
         db.write_execution_start(start)
@@ -151,9 +156,9 @@ class HistoryReadTests(unittest.TestCase):
             )
         self.require_reads(db)
         with db.recorded_history_snapshot():
-            # Outer whitespace joins the stored form's scope (#438); case and
-            # Unicode normalization stay distinct.
-            expected = (4, 1, 4, 4, 1, 1, 2, 2)
+            # Outer whitespace (#438) and case (#460) join one scope;
+            # Unicode normalization stays distinct.
+            expected = (5, 5, 5, 5, 1, 1, 2, 2)
             for i, count in enumerate(expected, 1):
                 self.assertEqual(db.list_history_window(i)["recorded_run_count"], count)
             self.assertIsNone(db.list_history_window(100))
@@ -1047,3 +1052,70 @@ class HistoryReadTests(unittest.TestCase):
         with db.recorded_history_snapshot():
             found = db.get_newest_pair_run(positive, pair)
         self.assertEqual(found["run"]["run_id"], newest.run_id)
+
+    def test_case_variant_raw_text_joins_the_saved_prompt_scope(self):
+        # #460: the hash folds case, so these runs count toward this prompt.
+        db = self.database()
+        positive = self.hashed(db, "\u00e9t\u00e9 lighthouse \u03c3")
+        raws = (
+            "\u00c9T\u00c9 Lighthouse \u03a3\n",
+            "\u00e9t\u00e9 lighthouse \u03c3",
+            " \u00c9t\u00e9 LIGHTHOUSE \u03a3",
+        )
+        starts = [
+            self.capture(db, i, bundle(positive=raw), time=100 + i)
+            for i, raw in enumerate(raws)
+        ]
+        # Unicode normalization and inner text stay distinct
+        self.capture(db, 7, bundle(positive="e\u0301te\u0301 lighthouse \u03c3"))
+        self.capture(db, 8, bundle(positive="\u00e9t\u00e9  lighthouse \u03c3"))
+        dotted = self.hashed(db, "\u0130stanbul")
+        dotted_run = self.capture(db, 9, bundle(positive="\u0130STANBUL"), time=300)
+        with db.recorded_history_snapshot():
+            page = db.list_history_window(positive)
+            details = [db.get_history_run(start.run_id) for start in starts]
+            turkish = db.list_history_window(dotted)
+        self.assertEqual(page["recorded_run_count"], 3)
+        self.assertEqual(
+            [r["run_id"] for r in page["items"]], [s.run_id for s in starts]
+        )
+        for raw, detail in zip(raws, details):
+            context = detail["contexts"][0]
+            self.assertEqual(context["positive"]["raw"], raw)
+            self.assertEqual(
+                context["positive_raw_text_id"], page["group"]["raw_text_id"]
+            )
+            self.assertEqual(context["positive_prompt_id"], positive)
+        self.assertEqual([r["run_id"] for r in turkish["items"]], [dotted_run.run_id])
+
+    def test_newest_pair_run_follows_case_variant_raw_text(self):
+        db = self.database()
+        pair = db.save_prompt_pair("a lighthouse", "fog")
+        positive = db.get_prompt_pair(pair)["positive_prompt_id"]
+        newest = self.capture(db, 0, bundle("FOG", positive="A Lighthouse"), time=100)
+        with db.recorded_history_snapshot():
+            found = db.get_newest_pair_run(positive, pair)
+        self.assertEqual(found["run"]["run_id"], newest.run_id)
+
+    def test_a_position_from_before_460_resumes_the_same_scope(self):
+        db = self.database()
+        positive = self.hashed(db, "A Lighthouse")
+        for i in range(3):
+            self.capture(db, i, bundle(positive="a lighthouse"), time=100 + i)
+        with db.recorded_history_snapshot():
+            first = db.list_history_window(positive, limit=2)
+            position = first["next_position"]
+            legacy = replace(
+                position, raw_text_id=content_digest("text", ("A Lighthouse",))
+            )
+            rest = db.list_history_window(positive, limit=2, after=legacy)
+            with self.assertRaises(ValueError):
+                db.list_history_window(
+                    positive,
+                    after=replace(position, raw_text_id=content_digest("text", ("x",))),
+                )
+        self.assertEqual(first["recorded_run_count"], 3)
+        self.assertEqual(len(rest["items"]), 1)
+        self.assertEqual(
+            rest["group"]["raw_text_id"], content_digest("text", ("a lighthouse",))
+        )

utils/hashing.py

@@ -58,11 +58,7 @@ def generate_prompt_hash(text: str) -> str:
     if not isinstance(text, str):
         raise TypeError("Text must be a string")
 
-    # Normalize the text by stripping whitespace and converting to lowercase
-    # for consistent hashing regardless of minor formatting differences
-    normalized_text = text.strip().lower()
-
-    return hashlib.sha256(normalized_text.encode("utf-8")).hexdigest()
+    return hashlib.sha256(prompt_key(text).encode("utf-8")).hexdigest()
 
 
 def stored_prompt_text(text: str) -> str:
@@ -93,6 +89,34 @@ STRIPPED_CODE_POINTS = (
 _STRIP_CODES = ",".join(map(str, STRIPPED_CODE_POINTS))
 
 
+def prompt_key(text: str) -> str:
+    """The dedup key: the stored text with case folded by ``str.lower()``.
+
+    Two texts with one key are one prompt (the hash is sha256 of the key), so
+    a run belongs to the history of the prompt with its key (#460).
+    """
+    return stored_prompt_text(text).lower()
+
+
+def _sql_key(value):
+    return prompt_key(value) if isinstance(value, str) else None
+
+
+def _sql_hash(value):
+    return generate_prompt_hash(value) if isinstance(value, str) else None
+
+
+def register_prompt_functions(conn) -> None:
+    """Give ``conn`` ``pm_prompt_key(x)`` and ``pm_prompt_hash(x)`` (#460).
+
+    SQLite's lower() folds ASCII only; these call Python, so SQL gets exactly
+    ``prompt_key`` / ``generate_prompt_hash`` (NULL for a non-text value).
+    Never used in schema objects, so the file stays readable without them.
+    """
+    conn.create_function("pm_prompt_key", 1, _sql_key, deterministic=True)
+    conn.create_function("pm_prompt_hash", 1, _sql_hash, deterministic=True)
+
+
 def stored_prompt_text_sql(expression: str) -> str:
     """``stored_prompt_text`` as an SQLite expression over ``expression``.