feat(history): expose fenced history windows and run details (#209)

2026-10-18 10:50:14 -07:00
parent f9f54c867a
commit e7c45ca946
8 changed files with 1400 additions and 3 deletions
@@ -87,6 +87,9 @@ try:
    routes = config.routes
    api = PromptManagerAPI()
    api.add_routes(routes)
+    host_app = getattr(config.server_instance, "app", None)
+    if host_app is not None and hasattr(host_app, "on_cleanup"):
+        host_app.on_cleanup.append(api.cleanup_history_reads)

 except Exception as e:
    # Log to internal logging system without console spam
@@ -24,6 +24,7 @@ from PIL import Image
 from .prompts import PromptRoutesMixin
 from .pairs import PairRoutesMixin
 from .library import LibraryRoutesMixin
+from .history import HistoryRoutesMixin
 from .images import ImageRoutesMixin
 from .admin import AdminRoutesMixin
 from .logging_routes import LoggingRoutesMixin
@@ -245,6 +246,7 @@ async def _gzip_middleware(request, handler):
 class PromptManagerAPI(
    PairRoutesMixin,
    LibraryRoutesMixin,
+    HistoryRoutesMixin,
    PromptRoutesMixin,
    ImageRoutesMixin,
    AdminRoutesMixin,
@@ -273,6 +275,8 @@ class PromptManagerAPI(
        self.logger.info("Initializing PromptManager API")

        self.db = db if db is not None else PromptDatabase()
+        self._history_read_fence = None
+        self._history_reads_closed = False
        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)
@@ -402,6 +406,7 @@ class PromptManagerAPI(

        self._register_pair_routes(routes)
        self._register_library_routes(routes)
+        self._register_history_routes(routes)
        self._register_prompt_routes(routes)
        self._register_image_routes(routes)
        self._register_admin_routes(routes)
@@ -0,0 +1,320 @@
+"""Authenticated chronological windows and bounded frozen run details."""
+
+import base64
+import binascii
+from functools import wraps
+import json
+import re
+import sqlite3
+from typing import Callable
+
+from aiohttp import web
+
+from .library import _decode, _encode, _scope
+
+try:
+    from ...database.execution_history import HistoryEpochChanged
+    from ...database.history_reads import RunKey, WindowPosition
+    from ...utils.history_read_fence import HistoryReadFence
+except ImportError:
+    from database.execution_history import HistoryEpochChanged
+    from database.history_reads import RunKey, WindowPosition
+    from utils.history_read_fence import HistoryReadFence
+
+
+_RUN_MAX_BYTES = 8 * 1024 * 1024
+_UUID = re.compile(r"[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12}")
+_DIGEST = re.compile(r"[0-9a-f]{64}")
+_DECIMAL = re.compile(r"0|[1-9][0-9]{0,18}")
+_RUN_DECIMALS = (
+    "queue_item_id",
+    "observed_dequeue_at",
+    "observed_callback_at",
+    "executor_started_at",
+    "terminal_event_at",
+)
+_MESSAGES = {
+    "invalid-request": "Invalid history request",
+    "invalid-cursor": "Invalid history cursor; refresh recorded history",
+    "not-found": "Recorded history target not found",
+    "history-changed": "Recorded history changed; reload recorded history",
+    "storage-unavailable": "Recorded history is temporarily unavailable",
+    "response-limit": "Recorded history response exceeds the response limit",
+}
+
+
+class HistoryReadError(Exception):
+    """A bounded public failure without private storage or capture details."""
+
+    def __init__(self, status: int, code: str, recovery: dict | None = None) -> None:
+        self.status = status
+        self.code = code
+        self.recovery = recovery
+        super().__init__(_MESSAGES[code])
+
+
+def _error_response(error: HistoryReadError) -> web.Response:
+    return web.json_response(
+        {
+            "success": False,
+            "history_version": 1,
+            "code": error.code,
+            "error": str(error),
+            "recovery": error.recovery,
+        },
+        status=error.status,
+        headers={"Cache-Control": "no-store"},
+    )
+
+
+def _history_errors(handler: Callable) -> Callable:
+    @wraps(handler)
+    async def wrapped(self, request: web.Request) -> web.Response:
+        try:
+            return await handler(self, request)
+        except HistoryReadError as error:
+            return _error_response(error)
+        except HistoryEpochChanged:
+            return _error_response(
+                HistoryReadError(409, "history-changed", {"action": "refresh-history"})
+            )
+        except ValueError:
+            return _error_response(HistoryReadError(400, "invalid-request"))
+        except (sqlite3.Error, OSError) as error:
+            self.logger.warning("History read unavailable (%s)", type(error).__name__)
+            return _error_response(
+                HistoryReadError(503, "storage-unavailable", {"action": "retry-read"})
+            )
+
+    return wrapped
+
+
+def _decimal(value: object, *, minimum: int = 0) -> int:
+    if type(value) is not str or _DECIMAL.fullmatch(value) is None:
+        raise ValueError("Invalid decimal history integer")
+    parsed = int(value)
+    if not minimum <= parsed <= 2**63 - 1:
+        raise ValueError("History integer outside supported range")
+    return parsed
+
+
+def _uuid(value: object) -> str:
+    if type(value) is not str or _UUID.fullmatch(value) is None:
+        raise ValueError("Invalid history UUID")
+    return value
+
+
+def _query(request: web.Request) -> None:
+    if any(len(request.query.getall(key)) != 1 for key in request.query):
+        raise ValueError("Duplicate history query parameter")
+
+
+def _page_limit(request: web.Request) -> int:
+    limit = _decimal(request.query.get("limit", "40"), minimum=1)
+    if limit > 100:
+        raise ValueError("History limit outside supported range")
+    return limit
+
+
+def _decode_run_key(value: object) -> RunKey:
+    if type(value) is not list or len(value) != 4:
+        raise ValueError("Invalid history run key")
+    return (_decimal(value[0]), _uuid(value[1]), _decimal(value[2]), _uuid(value[3]))
+
+
+def _wire_run_key(key: RunKey) -> list[str]:
+    return [str(key[0]), key[1], str(key[2]), key[3]]
+
+
+def _cursor_payload(token: str, kind: str) -> object:
+    """Use Library authentication, allowing only inline, domain-bound keys."""
+    try:
+        if not 1 <= len(token.encode("ascii")) <= 4096:
+            raise ValueError
+        raw = base64.b64decode(
+            token + "=" * (-len(token) % 4), altchars=b"-_", validate=True
+        )
+        wrapper = json.loads(raw[32:])
+        if type(wrapper) is not dict or set(wrapper) != {"scope", "key"}:
+            raise ValueError
+        return _decode(token, _scope({"kind": kind}))
+    except (ValueError, TypeError, binascii.Error, UnicodeError, RecursionError):
+        raise HistoryReadError(400, "invalid-cursor") from None
+
+
+def _decode_window_cursor(
+    token: str | None, positive_id: int
+) -> tuple[str | None, WindowPosition | None]:
+    if token is None:
+        return None, None
+    try:
+        value = _cursor_payload(token, "history-window")
+        if type(value) is not dict or set(value) != {
+            "version",
+            "epoch",
+            "positive_id",
+            "raw_text_id",
+            "direction",
+            "upper",
+            "before",
+        }:
+            raise ValueError
+        if (
+            type(value["version"]) is not int
+            or value["version"] != 1
+            or _decimal(value["positive_id"], minimum=1) != positive_id
+            or value["direction"] != "older"
+            or type(value["raw_text_id"]) is not str
+            or _DIGEST.fullmatch(value["raw_text_id"]) is None
+        ):
+            raise ValueError
+        epoch = _uuid(value["epoch"])
+        upper = _decode_run_key(value["upper"])
+        before = _decode_run_key(value["before"])
+        if before > upper:
+            raise ValueError
+        return epoch, WindowPosition(value["raw_text_id"], upper, before)
+    except ValueError:
+        raise HistoryReadError(400, "invalid-cursor") from None
+
+
+def _encode_window_cursor(
+    position: WindowPosition | None, epoch: str, positive_id: int
+) -> str | None:
+    if position is None:
+        return None
+    return _encode(
+        {
+            "version": 1,
+            "epoch": epoch,
+            "positive_id": str(positive_id),
+            "raw_text_id": position.raw_text_id,
+            "direction": "older",
+            "upper": _wire_run_key(position.upper),
+            "before": _wire_run_key(position.before),
+        },
+        _scope({"kind": "history-window"}),
+    )
+
+
+def _wire_run(run: dict) -> None:
+    for key in _RUN_DECIMALS:
+        if run[key] is not None:
+            run[key] = str(run[key])
+
+
+def _serialize_history(payload: dict, *, max_bytes: int | None = None) -> bytes:
+    try:
+        encoded = json.dumps(
+            payload, ensure_ascii=False, allow_nan=False, separators=(",", ":")
+        ).encode("utf-8")
+    except (ValueError, TypeError, UnicodeError):
+        raise sqlite3.DatabaseError("Invalid recorded history payload") from None
+    if max_bytes is not None and len(encoded) > max_bytes:
+        raise HistoryReadError(422, "response-limit")
+    return encoded
+
+
+def _success_response(encoded: bytes) -> web.Response:
+    return web.Response(
+        body=encoded,
+        content_type="application/json",
+        headers={"Cache-Control": "no-store"},
+    )
+
+
+class HistoryRoutesMixin:
+    def _register_history_routes(self, routes: web.RouteTableDef) -> None:
+        routes.get("/prompt_manager/library/{positive_id}/history")(
+            self.get_library_history
+        )
+        routes.get("/prompt_manager/history/runs/{run_id}")(self.get_history_run)
+
+    def _history_fence(self) -> HistoryReadFence:
+        # Called synchronously by the event-loop owner before scheduling work.
+        if self._history_reads_closed:
+            raise sqlite3.OperationalError("History reads are closed")
+        if self._history_read_fence is None:
+            self._history_read_fence = HistoryReadFence(self.db)
+        return self._history_read_fence
+
+    def close_history_reads(self) -> None:
+        self._history_reads_closed = True
+        if self._history_read_fence is not None:
+            self._history_read_fence.close()
+
+    async def cleanup_history_reads(self, app: web.Application) -> None:
+        self.close_history_reads()
+
+    @_history_errors
+    async def get_library_history(self, request: web.Request) -> web.Response:
+        _query(request)
+        positive_id = _decimal(request.match_info["positive_id"], minimum=1)
+        limit = _page_limit(request)
+        expected_epoch, position = _decode_window_cursor(
+            request.query.get("cursor"), positive_id
+        )
+        fence = self._history_fence()
+        db = self.db
+
+        def project() -> bytes:
+            generation = fence.begin()
+            with db.recorded_history_snapshot(expected_epoch) as epoch:
+                try:
+                    result = db.list_history_window(
+                        positive_id, limit=limit, after=position
+                    )
+                except ValueError:
+                    raise HistoryReadError(400, "invalid-cursor") from None
+            if result is None:
+                fence.validate(epoch, generation)
+                raise HistoryReadError(404, "not-found")
+            result["next_cursor"] = _encode_window_cursor(
+                result.pop("next_position"), epoch, positive_id
+            )
+            for run in result["items"]:
+                _wire_run(run)
+            encoded = _serialize_history(
+                {
+                    "success": True,
+                    "history_version": 1,
+                    "history_epoch": epoch,
+                    **result,
+                }
+            )
+            fence.validate(epoch, generation)
+            return encoded
+
+        return _success_response(await self._run_in_executor(project))
+
+    @_history_errors
+    async def get_history_run(self, request: web.Request) -> web.Response:
+        _query(request)
+        run_id = _uuid(request.match_info["run_id"])
+        expected_epoch = request.query.get("epoch")
+        if expected_epoch is not None:
+            _uuid(expected_epoch)
+        fence = self._history_fence()
+        db = self.db
+
+        def project() -> bytes:
+            generation = fence.begin()
+            with db.recorded_history_snapshot(expected_epoch) as epoch:
+                result = db.get_history_run(run_id)
+            if result is None:
+                fence.validate(epoch, generation)
+                raise HistoryReadError(404, "not-found")
+            _wire_run(result["run"])
+            encoded = _serialize_history(
+                {
+                    "success": True,
+                    "history_version": 1,
+                    "history_epoch": epoch,
+                    **result,
+                },
+                max_bytes=_RUN_MAX_BYTES,
+            )
+            fence.validate(epoch, generation)
+            return encoded
+
+        return _success_response(await self._run_in_executor(project))
@@ -222,3 +222,34 @@ test("latestOnly: errors from the current call propagate, errors from superseded
    rejectFirst(new Error("old failure"));
    assert.equal(await first, STALE);
 });
+
+test("ApiError exposes only validated history recovery actions on all error paths", async () => {
+    const permitted = [null, { action: "refresh-history" }, { action: "reload-outputs" },
+        { action: "retry-read" }, { action: "smaller-limit", limit: 1 },
+        { action: "smaller-limit", limit: 100 }];
+    const denied = [undefined, false, "refresh-history", [], {}, { action: "unknown" },
+        { action: "refresh-history", private: "secret" }, { action: "retry-read", limit: 1 },
+        { action: "smaller-limit" }, { action: "smaller-limit", limit: 0 },
+        { action: "smaller-limit", limit: 101 }, { action: "smaller-limit", limit: true },
+        { action: "smaller-limit", limit: "1" }, { action: "smaller-limit", limit: 1.5 },
+        { action: "smaller-limit", limit: NaN }, { action: "smaller-limit", limit: Infinity },
+        { action: "smaller-limit", limit: 10, url: "/private" },
+        Object.assign(Object.create({ action: "retry-read" }), { private: "secret" })];
+    assert.equal(new ApiError(409, "old caller").recovery, null);
+    for (const recovery of permitted.concat(denied)) {
+        for (const status of [409, 200]) {
+            const api = createApiClient(async () => jsonResponse(status, {
+                success: false, error: "bounded", code: "history-changed", recovery,
+            }));
+            await assert.rejects(api.get("/x"), error => {
+                assert.ok(error instanceof ApiError);
+                assert.equal(error.code, "history-changed");
+                assert.deepEqual(error.recovery, permitted.includes(recovery) ? recovery : null);
+                if (recovery && permitted.includes(recovery)) {
+                    assert.notEqual(error.recovery, recovery, "server objects must be copied");
+                }
+                return true;
+            });
+        }
+    }
+});
@@ -0,0 +1,744 @@
+"""History HTTP contracts use disposable SQLite, exact cursors and restore fences."""
+
+import base64
+import copy
+import importlib
+import json
+import sqlite3
+import tempfile
+import unittest
+from dataclasses import replace
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import patch
+
+from aiohttp import FormData, web
+
+try:
+    from test_api_images import ImageAPITestCase
+except ImportError:
+    from tests.test_api_images import ImageAPITestCase
+
+from database.operations import PromptDatabase
+from py.api.library import _encode, _scope
+from tests.test_execution_context_database import bundle, snapshot
+from tests.test_execution_history_database import JOB, SESSION
+from utils.execution_contexts import ContextBundle, RecipeContext, TextEvidence
+from utils.execution_history_events import make_finish, make_start
+
+OTHER_SESSION = "aaaaaaaa-3333-4333-8333-333333333333"
+MISSING_RUN = "bbbbbbbb-4444-4444-8444-444444444444"
+
+
+class HistoryAPITests(ImageAPITestCase):
+    async def tearDownAsync(self):
+        if hasattr(self.api, "close_history_reads"):
+            self.api.close_history_reads()
+        await super().tearDownAsync()
+
+    def capture(
+        self, item=0, time=100, session=SESSION, context=True, negative="negative"
+    ):
+        start = make_start(
+            session,
+            ((0, JOB, {}, {}), item),
+            self.api.db.get_execution_epoch(),
+            time,
+        )
+        self.api.db.write_execution_start(start)
+        self.api.db.write_execution_finish(
+            make_finish(
+                session, item, start.storage_epoch, SimpleNamespace(messages=[]), time
+            )
+        )
+        if context:
+            self.api.db.write_execution_contexts(
+                start.run_id, start.storage_epoch, bundle(negative)
+            )
+        return start
+
+    async def window(self, positive, params=None):
+        response = await self.client.get(
+            f"/prompt_manager/library/{positive}/history", params=params
+        )
+        self.assertEqual(
+            response.content_type, "application/json", "history route missing"
+        )
+        return response, await response.json()
+
+    async def detail(self, run, params=None):
+        response = await self.client.get(
+            f"/prompt_manager/history/runs/{run}", params=params
+        )
+        self.assertEqual(
+            response.content_type, "application/json", "history route missing"
+        )
+        return response, await response.json()
+
+    def assert_error(self, response, body, status, code, recovery=None):
+        self.assertEqual(response.status, status, body)
+        self.assertEqual(
+            body,
+            {
+                "success": False,
+                "history_version": 1,
+                "code": code,
+                "error": body["error"],
+                "recovery": recovery,
+            },
+        )
+        self.assertIsInstance(body["error"], str)
+        self.assertLessEqual(len(body["error"]), 160)
+        self.assertEqual(response.headers.get("Cache-Control"), "no-store")
+        self.assertNotIn(str(self.comfy_dir), json.dumps(body))
+
+    def cursor_payload(self, token):
+        decoded = base64.urlsafe_b64decode(token + "=" * (-len(token) % 4))
+        return json.loads(decoded[32:])["key"]
+
+    def signed(self, payload, kind="history-window"):
+        return _encode(payload, _scope({"kind": kind}))
+
+    async def test_window_defaults_bounds_and_no_context_recomposition(self):
+        positive = self._save_prompt("positive")
+        starts = [self.capture(i, i) for i in range(42)]
+        with patch.object(
+            self.api.db,
+            "get_execution_contexts",
+            side_effect=AssertionError("full contexts"),
+        ):
+            response, body = await self.window(positive)
+        self.assertEqual(response.status, 200, body)
+        self.assertEqual(body["history_version"], 1)
+        self.assertEqual(body["history_epoch"], starts[0].storage_epoch)
+        self.assertEqual(
+            [r["run_id"] for r in body["items"]], [s.run_id for s in starts[2:]]
+        )
+        self.assertEqual(body["recorded_run_count"], 42)
+        self.assertEqual(body["group"]["relation"], "exact_frozen_raw_text")
+        self.assertEqual(len(body["context_summaries"]), 1)
+        self.assertTrue(body["has_more"])
+        self.assertIsInstance(body["next_cursor"], str)
+        self.assertLessEqual(len(body["next_cursor"].encode()), 4096)
+        self.assertNotIn("next_position", body)
+        for limit, count in ((1, 1), (100, 42)):
+            response, page = await self.window(positive, {"limit": str(limit)})
+            self.assertEqual(response.status, 200)
+            self.assertEqual(len(page["items"]), count)
+        for params in (
+            [("limit", "1"), ("limit", "2")],
+            {"limit": "0"},
+            {"limit": "101"},
+            {"limit": "true"},
+            {"limit": "false"},
+            {"limit": "1.0"},
+            {"limit": "01"},
+            {"limit": "-1"},
+            {"limit": ""},
+            [("cursor", "a"), ("cursor", "b")],
+        ):
+            response, error = await self.window(positive, params)
+            self.assert_error(response, error, 400, "invalid-request")
+
+    async def test_exact_detail_preserves_frozen_roles_nulls_evidence_and_navigation(
+        self,
+    ):
+        pair = self.api.db.save_prompt_pair("positive", "negative")
+        positive = self.api.db.get_prompt_pair(pair)["positive_prompt_id"]
+        start = self.capture(9007199254740993, 2**63 - 1)
+        response, body = await self.detail(start.run_id, {"epoch": start.storage_epoch})
+        self.assertEqual(response.status, 200, body)
+        self.assertEqual(body["run"]["queue_item_id"], "9007199254740993")
+        self.assertEqual(body["run"]["observed_dequeue_at"], "9223372036854775807")
+        self.assertEqual(body["run"]["observed_callback_at"], "9223372036854775807")
+        self.assertIsNone(body["run"]["executor_started_at"])
+        context = body["contexts"][0]
+        self.assertEqual(
+            context["positive"],
+            {"raw": "positive", "effective": "positive", "prepend": "", "append": ""},
+        )
+        self.assertEqual(
+            context["negative"],
+            {"raw": "negative", "effective": "negative", "prepend": "", "append": ""},
+        )
+        self.assertEqual(context["reference_epoch"], start.storage_epoch)
+        self.assertEqual(
+            context["navigation"],
+            {
+                "positive_available": True,
+                "negative_available": True,
+                "pair_available": True,
+            },
+        )
+        for key in ("positive_snapshot_id", "negative_snapshot_id", "context_id"):
+            self.assertRegex(context[key], r"^[0-9a-f]{64}$")
+        self.assertEqual(
+            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,))
+        _, changed = await self.detail(start.run_id)
+        self.assertEqual(changed["contexts"][0]["positive"], context["positive"])
+        self.assertFalse(changed["contexts"][0]["navigation"]["positive_available"])
+        self.assertFalse(changed["contexts"][0]["navigation"]["pair_available"])
+        bare = self.capture(1, context=False)
+        response, uncaptured = await self.detail(bare.run_id)
+        self.assertEqual(response.status, 200)
+        self.assertEqual(uncaptured["context_capture"]["state"], "not_captured")
+        self.assertEqual(uncaptured["contexts"], [])
+        self.assertIsNone(uncaptured["output_summary"]["retained_occurrence_count"])
+
+    async def test_missing_and_invalid_ids_epochs_and_duplicate_queries(self):
+        response, body = await self.window(999999)
+        self.assert_error(response, body, 404, "not-found")
+        response, body = await self.detail(MISSING_RUN)
+        self.assert_error(response, body, 404, "not-found")
+        for identity in ("0", "-1", "true", "9223372036854775808", "01"):
+            response, body = await self.window(identity)
+            self.assert_error(response, body, 400, "invalid-request")
+        for run in ("not-a-uuid", MISSING_RUN.upper()):
+            response, body = await self.detail(run)
+            self.assert_error(response, body, 400, "invalid-request")
+        start = self.capture()
+        for params in (
+            {"epoch": ""},
+            {"epoch": "not-a-uuid"},
+            [("epoch", start.storage_epoch), ("epoch", start.storage_epoch)],
+        ):
+            response, body = await self.detail(start.run_id, params)
+            self.assert_error(response, body, 400, "invalid-request")
+        response, body = await self.detail(start.run_id, {"epoch": OTHER_SESSION})
+        self.assert_error(
+            response, body, 409, "history-changed", {"action": "refresh-history"}
+        )
+
+    async def test_nullable_frozen_fields_and_every_recorded_time_remain_exact(self):
+        start = make_start(
+            SESSION,
+            ((0, JOB, {}, {}), 9007199254740993),
+            self.api.db.get_execution_epoch(),
+            9007199254740993,
+        )
+        self.api.db.write_execution_start(start)
+        nullable = replace(
+            snapshot("positive"),
+            effective=TextEvidence(None, "unknown", "unknown", "missing"),
+        )
+        self.api.db.write_execution_contexts(
+            start.run_id,
+            start.storage_epoch,
+            ContextBundle(
+                (RecipeContext("7", nullable, None, "inferred"),),
+                "partial",
+                "captured",
+            ),
+        )
+        self.api.db.write_execution_finish(
+            make_finish(
+                SESSION,
+                start.queue_item_id,
+                start.storage_epoch,
+                SimpleNamespace(
+                    messages=[
+                        (
+                            "execution_start",
+                            {"prompt_id": JOB, "timestamp": 9007199254740993},
+                        ),
+                        (
+                            "execution_success",
+                            {"prompt_id": JOB, "timestamp": 2**63 - 1},
+                        ),
+                    ]
+                ),
+                2**63 - 1,
+            )
+        )
+        response, body = await self.detail(start.run_id)
+        self.assertEqual(response.status, 200)
+        self.assertEqual(body["run"]["executor_started_at"], "9007199254740993")
+        self.assertEqual(body["run"]["terminal_event_at"], "9223372036854775807")
+        self.assertEqual(body["run"]["observed_callback_at"], "9223372036854775807")
+        context = body["contexts"][0]
+        self.assertEqual(
+            context["positive"],
+            {"raw": "positive", "effective": None, "prepend": "", "append": ""},
+        )
+        self.assertIsNone(context["negative"])
+        self.assertIsNone(context["negative_snapshot_id"])
+        self.assertEqual(
+            context["evidence"]["positive"]["effective"]["provenance"], "unknown"
+        )
+
+    async def test_rf2_decimal_keysets_survive_deletion_and_delayed_capture(
+        self,
+    ):
+        positive = self._save_prompt("positive")
+        low = self.capture(0, 100)
+        delayed_old = self.capture(1, 100, context=False)
+        middle = self.capture(9007199254740993, 100)
+        deleted = self.capture(2**63 - 1, 100)
+        delayed_passed = self.capture(0, 100, OTHER_SESSION, context=False)
+        upper = self.capture(1, 100, OTHER_SESSION)
+        response, first = await self.window(positive, {"limit": "2"})
+        self.assertEqual(response.status, 200, first)
+        self.assertEqual(
+            [r["run_id"] for r in first["items"]], [deleted.run_id, upper.run_id]
+        )
+        self.assertEqual(first["items"][0]["queue_item_id"], "9223372036854775807")
+        self.assertEqual(first["recorded_run_count"], 4)
+        payload = self.cursor_payload(first["next_cursor"])
+        self.assertEqual(
+            set(payload),
+            {
+                "version",
+                "epoch",
+                "positive_id",
+                "raw_text_id",
+                "direction",
+                "upper",
+                "before",
+            },
+        )
+        self.assertEqual(payload["direction"], "older")
+        self.assertEqual(payload["positive_id"], str(positive))
+        self.assertEqual(
+            payload["before"], ["100", SESSION, "9223372036854775807", deleted.run_id]
+        )
+        with self.api.db.model.get_connection() as conn:
+            conn.execute(
+                "DELETE FROM execution_run_contexts WHERE run_id=?", (deleted.run_id,)
+            )
+            conn.execute(
+                "DELETE FROM execution_context_capture WHERE run_id=?",
+                (deleted.run_id,),
+            )
+            conn.execute("DELETE FROM execution_runs WHERE run_id=?", (deleted.run_id,))
+        newer = self.capture(7, 101)
+        for start in (delayed_old, delayed_passed):
+            self.api.db.write_execution_contexts(
+                start.run_id, start.storage_epoch, bundle()
+            )
+        _, second = await self.window(
+            positive, {"limit": "2", "cursor": first["next_cursor"]}
+        )
+        self.assertEqual(
+            [r["run_id"] for r in second["items"]], [delayed_old.run_id, middle.run_id]
+        )
+        self.assertEqual(second["items"][1]["queue_item_id"], "9007199254740993")
+        self.assertEqual(second["recorded_run_count"], 5)
+        _, last = await self.window(
+            positive, {"limit": "2", "cursor": second["next_cursor"]}
+        )
+        self.assertEqual([r["run_id"] for r in last["items"]], [low.run_id])
+        self.assertFalse(last["has_more"])
+        self.assertIsNone(last["next_cursor"])
+        pages = last["items"] + second["items"] + first["items"]
+        keys = [
+            (
+                int(r["observed_dequeue_at"]),
+                r["session_id"],
+                int(r["queue_item_id"]),
+                r["run_id"],
+            )
+            for r in pages
+        ]
+        self.assertEqual(keys, sorted(set(keys)))
+        _, refreshed = await self.window(positive)
+        self.assertEqual(refreshed["recorded_run_count"], 6)
+        self.assertIn(newer.run_id, [r["run_id"] for r in refreshed["items"]])
+        self.assertIn(delayed_passed.run_id, [r["run_id"] for r in refreshed["items"]])
+
+    async def test_cursor_authentication_strict_shape_and_domain_precede_projection(
+        self,
+    ):
+        positive = self._save_prompt("positive")
+        for i in range(3):
+            self.capture(i)
+        _, first = await self.window(positive, {"limit": "1"})
+        payload = self.cursor_payload(first["next_cursor"])
+        changed = []
+        for key, values in {
+            "version": [2, True, "1"],
+            "epoch": [OTHER_SESSION.upper(), "x"],
+            "positive_id": [str(positive + 1), positive, "01", True],
+            "raw_text_id": ["z" * 64, "A" * 64, None],
+            "direction": ["newer", None],
+            "upper": [
+                [],
+                ["1", SESSION, "1"],
+                ["01", SESSION, "1", MISSING_RUN],
+                [1, SESSION, "1", MISSING_RUN],
+                ["1", SESSION, "9223372036854775808", MISSING_RUN],
+                ["1", OTHER_SESSION.upper(), "1", MISSING_RUN],
+            ],
+            "before": [
+                False,
+                ["-1", SESSION, "0", MISSING_RUN],
+                ["1.0", SESSION, "0", MISSING_RUN],
+                ["1", SESSION, True, MISSING_RUN],
+            ],
+        }.items():
+            for value in values:
+                changed.append({**payload, key: value})
+        changed.extend(
+            [
+                {key: value for key, value in payload.items() if key != "before"},
+                {**payload, "private": "path"},
+            ]
+        )
+        tokens = [
+            "tampered",
+            "!",
+            "a" * 4097,
+            "猫",
+            self.signed(payload, "history-outputs"),
+            _encode(payload, _scope({"kind": "history-window"}), name=True),
+        ]
+        tokens += [self.signed(value) for value in changed]
+        token = first["next_cursor"]
+        tokens.append(("A" if token[0] != "A" else "B") + token[1:])
+        with patch.object(
+            self.api.db,
+            "list_history_window",
+            side_effect=AssertionError("projection must not run"),
+        ):
+            for token in tokens:
+                response, body = await self.window(positive, {"cursor": token})
+                self.assert_error(response, body, 400, "invalid-cursor")
+        with self.api.db.model.get_connection() as conn:
+            conn.execute("UPDATE prompts SET text='different' WHERE id=?", (positive,))
+        response, body = await self.window(positive, {"cursor": first["next_cursor"]})
+        self.assert_error(response, body, 400, "invalid-cursor")
+
+    async def test_valid_cursor_cannot_move_to_other_positive_or_raw_scope(self):
+        positive = self._save_prompt("positive")
+        duplicate = self._save_prompt("same saved row with another text")
+        with self.api.db.model.get_connection() as conn:
+            conn.execute("UPDATE prompts SET text='positive' WHERE id=?", (duplicate,))
+        self.capture(0)
+        self.capture(1)
+        _, first = await self.window(positive, {"limit": "1"})
+        with patch.object(
+            self.api.db, "list_history_window", side_effect=AssertionError("projection")
+        ):
+            response, body = await self.window(
+                duplicate, {"cursor": first["next_cursor"]}
+            )
+        self.assert_error(response, body, 400, "invalid-cursor")
+        payload = self.cursor_payload(first["next_cursor"])
+        response, body = await self.window(
+            positive, {"cursor": self.signed({**payload, "raw_text_id": "0" * 64})}
+        )
+        self.assert_error(response, body, 400, "invalid-cursor")
+
+    async def test_each_read_has_one_executor_job_and_ends_snapshot_before_validation(
+        self,
+    ):
+        positive = self._save_prompt("positive")
+        start = self.capture()
+        fence = self.api._history_fence()
+        validate = fence.validate
+        executor = self.api._run_in_executor
+        jobs = []
+        validated = []
+
+        def after_snapshot(epoch, generation):
+            self.assertFalse(self.api.db.model.get_connection().in_transaction)
+            validated.append(epoch)
+            validate(epoch, generation)
+
+        async def tracked_executor(func, *args, **kwargs):
+            jobs.append(func)
+            return await executor(func, *args, **kwargs)
+
+        with patch.object(fence, "validate", side_effect=after_snapshot):
+            with patch.object(
+                self.api, "_run_in_executor", side_effect=tracked_executor
+            ):
+                await self.window(positive)
+                await self.detail(start.run_id)
+        self.assertEqual(len(jobs), 2)
+        self.assertEqual(validated, [start.storage_epoch, start.storage_epoch])
+
+    async def test_final_storage_failure_and_invalid_serialization_cannot_succeed(self):
+        positive = self._save_prompt("positive")
+        start = self.capture()
+        fence = self.api._history_fence()
+        with patch.object(
+            fence, "validate", side_effect=sqlite3.OperationalError("private/path")
+        ):
+            for read in (
+                lambda: self.window(positive),
+                lambda: self.detail(start.run_id),
+            ):
+                response, body = await read()
+                self.assert_error(
+                    response, body, 503, "storage-unavailable", {"action": "retry-read"}
+                )
+        original = self.api.db.get_history_run
+
+        def invalid_payload(run_id):
+            result = original(run_id)
+            result["contexts"][0]["positive"]["raw"] = float("nan")
+            return result
+
+        with patch.object(self.api.db, "get_history_run", side_effect=invalid_payload):
+            response, body = await self.detail(start.run_id)
+        self.assert_error(
+            response, body, 503, "storage-unavailable", {"action": "retry-read"}
+        )
+
+    async def test_cursor_decoder_recursion_failure_is_bounded_before_projection(self):
+        positive = self._save_prompt("positive")
+        raw = (
+            b"x" * 32
+            + b'{"scope":"wrong","key":'
+            + b"[" * 1100
+            + b"0"
+            + b"]" * 1100
+            + b"}"
+        )
+        token = base64.urlsafe_b64encode(raw).decode().rstrip("=")
+        self.assertLess(len(token), 4096)
+        module = importlib.import_module("py.api.history")
+        original_loads = json.loads
+
+        def supported_runtime_loads(value, *args, **kwargs):
+            # Python 3.10's JSON decoder raises this for the actual deep token.
+            if type(value) is bytes and value == raw[32:]:
+                raise RecursionError("nested JSON")
+            return original_loads(value, *args, **kwargs)
+
+        with patch.object(
+            self.api.db, "list_history_window", side_effect=AssertionError("projection")
+        ):
+            with patch.object(
+                module.json, "loads", side_effect=supported_runtime_loads
+            ):
+                response, body = await self.window(positive, {"cursor": token})
+        self.assert_error(response, body, 400, "invalid-cursor")
+
+    async def test_epoch_change_is_409_but_tampering_remains_400(self):
+        positive = self._save_prompt("positive")
+        self.capture(0)
+        self.capture(1)
+        _, first = await self.window(positive, {"limit": "1"})
+        source = str(self.comfy_dir / "restore.db")
+        self.assertTrue(self.api.db.model.backup_database(source))
+        self.api.db.model.restore_from_file(source)
+        response, body = await self.window(positive, {"cursor": first["next_cursor"]})
+        self.assert_error(
+            response, body, 409, "history-changed", {"action": "refresh-history"}
+        )
+        response, body = await self.window(
+            positive, {"cursor": "bad" + first["next_cursor"]}
+        )
+        self.assert_error(response, body, 400, "invalid-cursor")
+
+    async def test_real_control_heavy_detail_uses_actual_utf8_budget_without_mutation(
+        self,
+    ):
+        self._save_prompt("positive")
+        negative = '猫😀\0\x01\n"\\' * 1500
+        start = self.capture(negative=negative)
+        response, body = await self.detail(start.run_id)
+        self.assertEqual(response.status, 200, body)
+        encoded = await response.read()
+        self.assertEqual(body["contexts"][0]["negative"]["raw"], negative)
+        self.assertIn("猫".encode(), encoded)
+        self.assertIn(b"\\u0000", encoded)
+        self.assertGreater(len(encoded), len(json.dumps(body, ensure_ascii=False)))
+        module = importlib.import_module("py.api.history")
+        with patch.object(module, "_RUN_MAX_BYTES", len(encoded)):
+            at, same = await self.detail(start.run_id)
+            self.assertEqual(at.status, 200)
+            self.assertEqual(await at.read(), encoded)
+            self.assertEqual(same, body)
+        with patch.object(module, "_RUN_MAX_BYTES", len(encoded) - 1):
+            above, error = await self.detail(start.run_id)
+            self.assert_error(above, error, 422, "response-limit")
+            self.assertNotIn("contexts", error)
+        _, after = await self.detail(start.run_id)
+        self.assertEqual(after, body)
+
+    async def test_projected_detail_fixture_pins_defensive_eight_mib_encoder_boundary(
+        self,
+    ):
+        start = self.capture()
+        response, detail = await self.detail(start.run_id)
+        self.assertEqual(response.status, 200, detail)
+
+        def check_projection_limit():
+            # A projected-detail fixture tests the defensive transport ceiling;
+            # valid stored captures retain their existing one-MiB text allowance.
+            projected = copy.deepcopy(detail)
+            projected["contexts"][0]["positive"]["raw"] = "猫😀\0"
+            baseline = json.dumps(
+                projected, ensure_ascii=False, allow_nan=False, separators=(",", ":")
+            ).encode()
+            projected["contexts"][0]["positive"]["raw"] += "x" * (
+                8 * 1024 * 1024 - len(baseline)
+            )
+            module = importlib.import_module("py.api.history")
+            encoded = module._serialize_history(projected, max_bytes=8 * 1024 * 1024)
+            self.assertEqual(len(encoded), 8 * 1024 * 1024)
+            self.assertEqual(json.loads(encoded), projected)
+            projected["contexts"][0]["positive"]["raw"] += "x"
+            with self.assertRaises(module.HistoryReadError) as error:
+                module._serialize_history(projected, max_bytes=8 * 1024 * 1024)
+            self.assertEqual(error.exception.code, "response-limit")
+
+        await self.api._run_in_executor(check_projection_limit)
+
+    async def test_storage_failure_is_bounded_retry_without_logging_private_values(
+        self,
+    ):
+        positive = self._save_prompt("positive")
+        start = self.capture()
+        private = "SECRET_PROMPT /private/data/secret.db"
+        for method, read in (
+            ("list_history_window", lambda: self.window(positive)),
+            ("get_history_run", lambda: self.detail(start.run_id)),
+        ):
+            for exc in (
+                sqlite3.OperationalError("busy " + private),
+                sqlite3.DatabaseError("corrupt " + private),
+                OSError(private),
+            ):
+                with patch.object(self.api.db, method, side_effect=exc):
+                    with self.assertLogs("prompt_manager.api", "WARNING") as logs:
+                        response, body = await read()
+                self.assert_error(
+                    response, body, 503, "storage-unavailable", {"action": "retry-read"}
+                )
+                self.assertNotIn(private, json.dumps(body) + " ".join(logs.output))
+
+    async def test_snapshot_restore_before_notification_rejects_both_public_reads(self):
+        positive = self._save_prompt("positive")
+        start = self.capture()
+        source = str(self.comfy_dir / "source.db")
+        self.assertTrue(self.api.db.model.backup_database(source))
+        writer = PromptDatabase(self.api.db.model.db_path)
+        self.addCleanup(writer.close_all)
+        for method, read in (
+            ("list_history_window", lambda: self.window(positive)),
+            ("get_history_run", lambda: self.detail(start.run_id)),
+        ):
+            original = getattr(self.api.db, method)
+
+            def during_snapshot(*args, **kwargs):
+                result = original(*args, **kwargs)
+                with patch("database.execution_history.notify_history_restore"):
+                    writer.model.restore_from_file(source)
+                return result
+
+            with patch.object(self.api.db, method, side_effect=during_snapshot):
+                response, body = await read()
+            self.assert_error(
+                response, body, 409, "history-changed", {"action": "refresh-history"}
+            )
+
+    async def test_one_lazy_subscription_cleanup_and_no_reinitialization(self):
+        import database.execution_history as observers
+
+        positive = self._save_prompt("positive")
+        start = self.capture()
+        identity = str(Path(self.api.db.model.db_path).resolve())
+        self.assertNotIn(identity, observers._restore_observers)
+        executor = self.api._run_in_executor
+
+        async def subscribed_before_executor(func, *args, **kwargs):
+            self.assertEqual(len(observers._restore_observers[identity]), 1)
+            return await executor(func, *args, **kwargs)
+
+        with patch.object(
+            self.api, "_run_in_executor", side_effect=subscribed_before_executor
+        ):
+            await self.window(positive)
+            await self.detail(start.run_id)
+            await self.window(positive)
+        self.assertEqual(len(observers._restore_observers[identity]), 1)
+        self.api.close_history_reads()
+        self.api.close_history_reads()
+        self.assertNotIn(identity, observers._restore_observers)
+        response, body = await self.detail(start.run_id)
+        self.assert_error(
+            response, body, 503, "storage-unavailable", {"action": "retry-read"}
+        )
+        self.assertNotIn(identity, observers._restore_observers)
+
+    async def test_cleanup_before_first_read_does_not_subscribe(self):
+        import database.execution_history as observers
+
+        identity = str(Path(self.api.db.model.db_path).resolve())
+        self.assertTrue(
+            callable(getattr(self.api, "close_history_reads", None)),
+            "API needs permanent history cleanup",
+        )
+        self.api.close_history_reads()
+        response, body = await self.window(self._save_prompt("positive"))
+        self.assert_error(
+            response, body, 503, "storage-unavailable", {"action": "retry-read"}
+        )
+        self.assertNotIn(identity, observers._restore_observers)
+
+    async def test_multipart_restore_then_read_uses_new_owner_and_current_epoch(self):
+        positive = self._save_prompt("positive")
+        start = self.capture()
+        _, before = await self.window(positive)
+        old_db = self.api.db
+        source = str(self.comfy_dir / "upload.db")
+        self.assertTrue(old_db.model.backup_database(source))
+        form = FormData()
+        form.add_field("database_file", Path(source).read_bytes(), filename="upload.db")
+        response = await self.client.post("/prompt_manager/restore", data=form)
+        self.assertEqual(response.status, 200, await response.text())
+        self.assertIsNot(self.api.db, old_db)
+        self.assertEqual(len(old_db.model._connections), 0)
+        with patch.object(
+            old_db.model, "get_connection", side_effect=AssertionError("retired owner")
+        ):
+            response, after = await self.window(positive)
+            self.assertEqual(response.status, 200, after)
+            response, detail = await self.detail(start.run_id)
+            self.assertEqual(response.status, 200, detail)
+        self.assertNotEqual(before["history_epoch"], after["history_epoch"])
+        self.assertEqual(detail["history_epoch"], after["history_epoch"])
+        self.assertEqual(detail["run"]["storage_epoch"], start.storage_epoch)
+        self.assertEqual(detail["contexts"][0]["reference_epoch"], start.storage_epoch)
+        self.assertEqual(
+            detail["contexts"][0]["navigation"],
+            {
+                "positive_available": False,
+                "negative_available": False,
+                "pair_available": False,
+            },
+        )
+        self.assertEqual(len(old_db.model._connections), 0)
+
+
+class HistoryHostCleanupTests(unittest.IsolatedAsyncioTestCase):
+    async def test_root_registers_cleanup_on_actual_host_application(self):
+        from tests.test_init_package import fake_server_module, load_package
+
+        with tempfile.TemporaryDirectory() as directory:
+            server = fake_server_module()
+            server.PromptServer.instance.app = web.Application()
+            with load_package(
+                "history_cleanup_pkg",
+                str(Path(directory) / "db.sqlite"),
+                directory,
+                server,
+            ) as (package, host):
+                self.assertIn(package.api.cleanup_history_reads, host.app.on_cleanup)
+                api = package.api
+                self.assertIsNone(getattr(api, "_history_read_fence", None))
+                fence = api._history_fence()
+                generation = fence.begin()
+                epoch = api.db.get_execution_epoch()
+                host.app.freeze()
+                await host.app.cleanup()
+                with self.assertRaises(sqlite3.Error):
+                    fence.validate(epoch, generation)
+                with self.assertRaises(sqlite3.Error):
+                    api._history_fence()
@@ -0,0 +1,202 @@
+"""Fresh epoch validation survives ended snapshots and retired API owners."""
+
+import importlib
+import sqlite3
+import unittest
+from contextlib import closing
+from pathlib import Path
+from unittest.mock import patch
+
+from database.execution_history import HistoryEpochChanged, notify_history_restore
+from tests import test_execution_history_database as history_tests
+
+
+class HistoryReadFenceTests(unittest.TestCase):
+    setUp = history_tests.HistoryDatabaseTests.setUp
+    close_databases = history_tests.HistoryDatabaseTests.close_databases
+    database = history_tests.HistoryDatabaseTests.database
+    start = history_tests.HistoryDatabaseTests.start
+
+    def fence(self, db):
+        try:
+            module = importlib.import_module("utils.history_read_fence")
+        except ImportError:
+            self.fail("HistoryReadFence must protect publication of history reads")
+        fence = module.HistoryReadFence(db)
+        self.addCleanup(fence.close)
+        return fence
+
+    def snapshot_file(self, db):
+        path = str(Path(self.temp.name) / "source.db")
+        self.assertTrue(db.model.backup_database(path))
+        return path
+
+    def test_final_epoch_is_fresh_after_snapshot_even_before_restore_notification(self):
+        reader = self.database()
+        writer = self.database()
+        start = self.start(reader)
+        reader.write_execution_start(start)
+        source = self.snapshot_file(reader)
+        fence = self.fence(reader)
+        generation = fence.begin()
+        with reader.recorded_history_snapshot() as epoch:
+            original = reader.get_history_run(start.run_id)
+            with patch("database.execution_history.notify_history_restore"):
+                writer.model.restore_from_file(source)
+            self.assertEqual(reader.get_execution_epoch(), epoch)
+            self.assertEqual(original["run"]["storage_epoch"], epoch)
+        with self.assertRaises(HistoryEpochChanged):
+            fence.validate(epoch, generation)
+        self.assertEqual(fence.begin(), generation)
+
+    def test_restore_observer_rejects_same_epoch_even_when_storage_did_not_change(self):
+        db = self.database()
+        fence = self.fence(db)
+        generation = fence.begin()
+        epoch = db.get_execution_epoch()
+        notify_history_restore(self.path)
+        with self.assertRaises(HistoryEpochChanged):
+            fence.validate(epoch, generation)
+        self.assertGreater(fence.begin(), generation)
+        fence.validate(epoch, fence.begin())
+
+    def test_new_read_before_delayed_notification_can_return_coherent_new_epoch(self):
+        db = self.database()
+        writer = self.database()
+        fence = self.fence(db)
+        old_epoch = db.get_execution_epoch()
+        source = self.snapshot_file(db)
+        with patch("database.execution_history.notify_history_restore"):
+            writer.model.restore_from_file(source)
+        generation = fence.begin()
+        with db.recorded_history_snapshot() as epoch:
+            self.assertNotEqual(epoch, old_epoch)
+            self.assertIsNone(
+                db.get_history_run("11111111-1111-4111-8111-111111111111")
+            )
+        fence.validate(epoch, generation)
+        notify_history_restore(self.path)
+        with self.assertRaises(HistoryEpochChanged):
+            fence.validate(epoch, generation)
+
+    def test_restore_after_final_validation_is_later_staleness_not_relabeling(self):
+        db = self.database()
+        writer = self.database()
+        start = self.start(db)
+        db.write_execution_start(start)
+        source = self.snapshot_file(db)
+        fence = self.fence(db)
+        generation = fence.begin()
+        with db.recorded_history_snapshot() as epoch:
+            detail = db.get_history_run(start.run_id)
+        fence.validate(epoch, generation)
+        writer.model.restore_from_file(source)
+        self.assertEqual(detail["run"]["storage_epoch"], epoch)
+        self.assertNotEqual(writer.get_execution_epoch(), epoch)
+
+    def test_validation_does_not_reopen_retired_owner_and_closes_read_only_handle(self):
+        db = self.database()
+        epoch = db.get_execution_epoch()
+        fence = self.fence(db)
+        generation = fence.begin()
+        db.close_all()
+        module = importlib.import_module("utils.history_read_fence")
+        connect = sqlite3.connect
+        opened = []
+
+        def tracked(*args, **kwargs):
+            conn = connect(*args, **kwargs)
+            with self.assertRaises(sqlite3.OperationalError):
+                conn.execute(
+                    "UPDATE execution_history_metadata SET storage_epoch=storage_epoch"
+                )
+            opened.append(conn)
+            return conn
+
+        with patch.object(
+            db.model, "get_connection", side_effect=AssertionError("retired")
+        ):
+            with patch.object(module.sqlite3, "connect", side_effect=tracked):
+                fence.validate(epoch, generation)
+        self.assertEqual(len(opened), 1)
+        with self.assertRaises(sqlite3.ProgrammingError):
+            opened[0].execute("SELECT 1")
+        self.assertEqual(len(db.model._connections), 0)
+
+    def test_observer_or_close_between_epoch_select_and_publish_rejects_read(self):
+        db = self.database()
+        module = importlib.import_module("utils.history_read_fence")
+        original_connect = sqlite3.connect
+        epoch = db.get_execution_epoch()
+        for close in (False, True):
+            fence = self.fence(db)
+            generation = fence.begin()
+
+            class InterruptedConnection(sqlite3.Connection):
+                def close(connection):
+                    super().close()
+                    if close:
+                        fence.close()
+                    else:
+                        notify_history_restore(self.path)
+
+            def interrupted_connect(*args, **kwargs):
+                return original_connect(*args, factory=InterruptedConnection, **kwargs)
+
+            with patch.object(
+                module.sqlite3, "connect", side_effect=interrupted_connect
+            ):
+                with self.assertRaises(sqlite3.Error if close else HistoryEpochChanged):
+                    fence.validate(epoch, generation)
+
+    def test_final_select_failure_always_closes_its_handle(self):
+        db = self.database()
+        fence = self.fence(db)
+        module = importlib.import_module("utils.history_read_fence")
+        epoch = db.get_execution_epoch()
+        with db.model.get_connection() as conn:
+            conn.execute("DROP TABLE execution_history_metadata")
+        original_connect = sqlite3.connect
+        opened = []
+
+        def tracked(*args, **kwargs):
+            conn = original_connect(*args, **kwargs)
+            opened.append(conn)
+            return conn
+
+        with patch.object(module.sqlite3, "connect", side_effect=tracked):
+            with self.assertRaises(sqlite3.DatabaseError):
+                fence.validate(epoch, fence.begin())
+        self.assertEqual(len(opened), 1)
+        with self.assertRaises(sqlite3.ProgrammingError):
+            opened[0].execute("SELECT 1")
+
+    def test_missing_metadata_and_connection_failure_are_not_empty_success(self):
+        db = self.database()
+        fence = self.fence(db)
+        generation = fence.begin()
+        epoch = db.get_execution_epoch()
+        with closing(sqlite3.connect(self.path)) as conn:
+            conn.execute("DELETE FROM execution_history_metadata")
+            conn.commit()
+        with self.assertRaises(sqlite3.DatabaseError):
+            fence.validate(epoch, generation)
+        module = importlib.import_module("utils.history_read_fence")
+        with patch.object(
+            module.sqlite3, "connect", side_effect=OSError("private/path")
+        ):
+            with self.assertRaises(OSError):
+                fence.validate(epoch, generation)
+
+    def test_close_is_idempotent_unsubscribes_and_permanently_invalidates_reads(self):
+        db = self.database()
+        fence = self.fence(db)
+        epoch = db.get_execution_epoch()
+        generation = fence.begin()
+        fence.close()
+        fence.close()
+        notify_history_restore(self.path)
+        with self.assertRaises(sqlite3.Error):
+            fence.begin()
+        with self.assertRaises(sqlite3.Error):
+            fence.validate(epoch, generation)
@@ -0,0 +1,80 @@
+"""Restore publication checks for snapshot-owned history reads."""
+
+import os
+from pathlib import Path
+import sqlite3
+import threading
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from ..database.operations import PromptDatabase
+
+try:
+    from ..database.execution_history import (
+        HistoryEpochChanged,
+        subscribe_history_restore,
+    )
+except ImportError:
+    from database.execution_history import (
+        HistoryEpochChanged,
+        subscribe_history_restore,
+    )
+
+
+class HistoryReadFence:
+    """Combine a fresh storage epoch with process-local restore notifications."""
+
+    def __init__(self, db: "PromptDatabase") -> None:
+        self._path = os.path.normcase(os.path.realpath(db.model.db_path))
+        self._lock = threading.Lock()
+        self._generation = 0
+        self._closed = False
+        self._unsubscribe = subscribe_history_restore(db.model.db_path, self._restored)
+
+    def _restored(self) -> None:
+        with self._lock:
+            if not self._closed:
+                self._generation += 1
+
+    def _check(self, generation: int | None = None, epoch: str = "") -> int:
+        if self._closed:
+            raise sqlite3.OperationalError("History reads are closed")
+        if generation is not None and generation != self._generation:
+            raise HistoryEpochChanged(epoch)
+        return self._generation
+
+    def begin(self) -> int:
+        with self._lock:
+            return self._check()
+
+    def validate(self, epoch: str, generation: int) -> None:
+        """Validate after the caller's read transaction and response work ended."""
+        with self._lock:
+            self._check(generation, epoch)
+        # An API restore retires its previous model. Never reopen that owner,
+        # and never reuse the projection transaction's view of the epoch.
+        conn = sqlite3.connect(
+            Path(self._path).as_uri() + "?mode=ro", uri=True, timeout=0.25
+        )
+        try:
+            row = conn.execute(
+                "SELECT storage_epoch FROM execution_history_metadata "
+                "WHERE singleton_id=1"
+            ).fetchone()
+            if row is None or not isinstance(row[0], str):
+                raise sqlite3.DatabaseError("Missing history storage epoch")
+            current_epoch = row[0]
+        finally:
+            conn.close()
+        with self._lock:
+            self._check(generation, epoch)
+            if current_epoch != epoch:
+                raise HistoryEpochChanged(current_epoch)
+
+    def close(self) -> None:
+        with self._lock:
+            if self._closed:
+                return
+            self._closed = True
+            self._generation += 1
+        self._unsubscribe()
@@ -15,11 +15,23 @@
    "use strict";

    class ApiError extends Error {
-        constructor(status, message, code = null) {
+        constructor(status, message, code = null, recovery = null) {
            super(message);
            this.name = "ApiError";
            this.status = status;
            this.code = typeof code === "string" && /^[a-z0-9-]{1,64}$/.test(code) ? code : null;
+            this.recovery = null;
+            if (recovery && typeof recovery === "object" && !Array.isArray(recovery)) {
+                const keys = Object.keys(recovery);
+                if (keys.length === 1 && keys[0] === "action" &&
+                    ["refresh-history", "reload-outputs", "retry-read"].includes(recovery.action)) {
+                    this.recovery = { action: recovery.action };
+                } else if (keys.length === 2 && keys.includes("action") && keys.includes("limit") &&
+                    recovery.action === "smaller-limit" && Number.isInteger(recovery.limit) &&
+                    recovery.limit >= 1 && recovery.limit <= 100) {
+                    this.recovery = { action: "smaller-limit", limit: recovery.limit };
+                }
+            }
        }
    }

@@ -63,11 +75,11 @@

            const serverMessage = data && typeof data === "object" ? data.error || data.message : undefined;
            if (!response.ok) {
-                throw new ApiError(response.status, serverMessage || response.statusText || `HTTP ${response.status}`, data?.code);
+                throw new ApiError(response.status, serverMessage || response.statusText || `HTTP ${response.status}`, data?.code, data?.recovery);
            }
            if (!parsed) throw new ApiError(response.status, "Invalid JSON response");
            if (data && typeof data === "object" && data.success === false) {
-                throw new ApiError(response.status, serverMessage || "Request failed", data.code);
+                throw new ApiError(response.status, serverMessage || "Request failed", data.code, data.recovery);
            }
            return data;
        }