ComfyAssets-ComfyUI_PromptManager/tests/test_api_history.py

"""History HTTP contracts use disposable SQLite, exact cursors and restore fences."""

import asyncio
import base64
import copy
import importlib
import json
import os
import sqlite3
import tempfile
import threading
import unittest
from contextlib import contextmanager, ExitStack
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

# Ceiling for the restore handoffs in assert_projection_releases_retired_owner.
# Each wait ends on an event (the read reaching its barrier, the test releasing
# it after a full multipart restore); the ceiling only stops a broken test from
# hanging. A slow Windows runner took longer than 5 s for that restore, so the
# blocked read raised inside its handler and returned a text/plain 500 (#242).
HANDOFF_TIMEOUT_SECONDS = 60

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]
        _, window = await self.window(positive)
        self.assertEqual(
            context["positive_raw_text_id"], window["group"]["raw_text_id"]
        )
        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_window_and_exact_run_publish_boolean_diagnostics_from_sqlite(
        self,
    ) -> None:
        positive = self._save_prompt("positive")
        starts = []
        for item, omitted in enumerate((False, True)):
            start = make_start(
                SESSION,
                ((0, JOB, {}, {}), item),
                self.api.db.get_execution_epoch(),
                item,
            )
            self.api.db.write_execution_start(start)
            finish = make_finish(
                SESSION,
                item,
                start.storage_epoch,
                SimpleNamespace(
                    messages=[
                        (
                            "execution_success",
                            {
                                "prompt_id": JOB,
                                "timestamp": item,
                                "node_type": object() if omitted else None,
                            },
                        )
                    ]
                ),
                item,
            )
            self.api.db.write_execution_finish(finish)
            self.api.db.write_execution_contexts(
                start.run_id, start.storage_epoch, bundle()
            )
            self.assertEqual(
                self.api.db.get_execution_run(start.run_id)["diagnostics_omitted"],
                int(omitted),
            )
            starts.append(start)
        response, window = await self.window(positive)
        self.assertEqual(response.status, 200, window)
        for start, omitted in zip(starts, (False, True)):
            with self.subTest(omitted=omitted):
                summary = next(
                    row for row in window["items"] if row["run_id"] == start.run_id
                )
                self.assertIs(summary["diagnostics_omitted"], omitted)
                response, detail = await self.detail(start.run_id)
                self.assertEqual(response.status, 200, detail)
                self.assertIs(detail["run"]["diagnostics_omitted"], omitted)

    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_newest_pair_run_route_names_the_run_or_none(self):
        pair = self.api.db.save_prompt_pair("positive", "negative")
        quiet = self.api.db.save_prompt_pair("positive", "quiet")
        foreign = self.api.db.save_prompt_pair("elsewhere", "negative")
        positive = self.api.db.get_prompt_pair(pair)["positive_prompt_id"]
        self.capture(0, 100)
        newest = self.capture(1, 200)
        base = f"/prompt_manager/library/{positive}/history/pairs"
        response = await self.client.get(f"{base}/{pair}/newest")
        body = await response.json()
        self.assertEqual(response.status, 200, body)
        self.assertEqual(response.headers.get("Cache-Control"), "no-store")
        self.assertEqual(body["history_version"], 1)
        self.assertEqual(body["history_epoch"], newest.storage_epoch)
        self.assertEqual(body["run"]["run_id"], newest.run_id)
        self.assertEqual(body["run"]["anchor_id"], "4")
        response = await self.client.get(f"{base}/{quiet}/newest")
        self.assertEqual((await response.json())["run"], None)
        response = await self.client.get(f"{base}/{foreign}/newest")
        self.assert_error(response, await response.json(), 404, "not-found")
        response = await self.client.get(
            f"/prompt_manager/library/999999/history/pairs/{pair}/newest"
        )
        self.assert_error(response, await response.json(), 404, "not-found")
        for path in (f"{base}/0/newest", f"{base}/01/newest", f"{base}/true/newest"):
            response = await self.client.get(path)
            self.assert_error(response, await response.json(), 400, "invalid-request")
        response = await self.client.get(f"{base}/{pair}/newest", params={"x": "1"})
        self.assert_error(response, await response.json(), 400, "invalid-request")

    async def test_newest_pair_run_route_reports_storage_and_epoch_failures(self):
        pair = self.api.db.save_prompt_pair("positive", "negative")
        saved = self.api.db.get_prompt_pair(pair)
        self.capture(0, 100)
        url = (
            f"/prompt_manager/library/{saved['positive_prompt_id']}"
            f"/history/pairs/{pair}/newest"
        )
        # A prompt saved only as a negative is not a Library positive.
        response = await self.client.get(
            f"/prompt_manager/library/{saved['negative_prompt_id']}"
            f"/history/pairs/{pair}/newest"
        )
        self.assert_error(response, await response.json(), 404, "not-found")
        private = "SECRET_PROMPT /private/data/secret.db"
        for exc in (sqlite3.OperationalError("busy " + private), OSError(private)):
            with patch.object(self.api.db, "get_newest_pair_run", side_effect=exc):
                with self.assertLogs("prompt_manager.api", "WARNING") as logs:
                    response = await self.client.get(url)
                body = await response.json()
            self.assert_error(
                response, body, 503, "storage-unavailable", {"action": "retry-read"}
            )
            self.assertNotIn(private, json.dumps(body) + " ".join(logs.output))
        source = str(self.comfy_dir / "source.db")
        self.assertTrue(self.api.db.model.backup_database(source))
        writer = PromptDatabase(self.api.db.model.db_path)
        original = self.api.db.get_newest_pair_run

        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

        try:
            with patch.object(
                self.api.db, "get_newest_pair_run", side_effect=during_snapshot
            ):
                response = await self.client.get(url)
            self.assert_error(
                response,
                await response.json(),
                409,
                "history-changed",
                {"action": "refresh-history"},
            )
        finally:
            writer.close_all()

    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_resume_rebuilds_the_old_bounded_range_after_many_new_runs(self):
        positive = self._save_prompt("positive")
        starts = [self.capture(i, i) for i in range(42)]
        _, first = await self.window(positive)
        self.assertIn("resume_cursor", first)
        self.assertIsInstance(first["resume_cursor"], str)
        self.assertLessEqual(len(first["resume_cursor"]), 4096)
        _, exhausted = await self.window(positive, {"cursor": first["next_cursor"]})
        self.assertFalse(exhausted["has_more"])
        self.assertIsNone(exhausted["next_cursor"])
        self.assertIsInstance(exhausted["resume_cursor"], str)
        for i in range(42, 103):
            self.capture(i, i)
        response, resumed = await self.window(
            positive, {"resume": exhausted["resume_cursor"]}
        )
        self.assertEqual(response.status, 200, resumed)
        self.assertEqual(resumed["recorded_run_count"], 42)
        self.assertEqual(
            [run["run_id"] for run in resumed["items"]],
            [start.run_id for start in starts[2:]],
        )
        self.assertEqual(resumed["resume_cursor"], first["resume_cursor"])
        response, older = await self.window(
            positive, {"cursor": resumed["next_cursor"]}
        )
        self.assertEqual(response.status, 200, older)
        self.assertEqual(
            [run["run_id"] for run in older["items"]],
            [start.run_id for start in starts[:2]],
        )
        _, fresh = await self.window(positive)
        self.assertEqual(fresh["recorded_run_count"], 103)
        self.assertTrue(
            set(run["run_id"] for run in fresh["items"]).isdisjoint(
                start.run_id for start in starts
            )
        )

    async def test_exhausted_resume_authenticates_scope_and_distinguishes_epoch_change(
        self,
    ):
        positive = self._save_prompt("positive")
        starts = [self.capture(i, i) for i in range(3)]
        _, first = await self.window(positive)
        self.assertIsNone(first["next_cursor"])
        self.assertIn("resume_cursor", first)
        token = first["resume_cursor"]
        response, resumed = await self.window(positive, {"resume": token})
        self.assertEqual(response.status, 200, resumed)
        self.assertEqual(
            [run["run_id"] for run in resumed["items"]],
            [start.run_id for start in starts],
        )
        _, ordinary = await self.window(positive, {"limit": "1"})
        for params in (
            {"resume": ""},
            {"resume": "x" * 4097},
            {"resume": "bad" + token},
            {"cursor": token},
            {"resume": ordinary["next_cursor"]},
        ):
            response, body = await self.window(positive, params)
            self.assert_error(response, body, 400, "invalid-cursor")
        for params in (
            {"resume": token, "cursor": ordinary["next_cursor"]},
            [("resume", token), ("resume", token)],
            {"resume": token, "limit": "101"},
            {"resume": token, "unknown": "value"},
        ):
            response, body = await self.window(positive, params)
            self.assert_error(response, body, 400, "invalid-request")
        payload = self.cursor_payload(token)
        for change in (
            {"version": True},
            {"positive_id": positive},
            {"positive_id": "999"},
            {"raw_text_id": "bad"},
            {"upper": ["01", SESSION, "0", starts[0].run_id]},
            {"upper": ["0", SESSION, True, starts[0].run_id]},
            {"upper": []},
            {"before": payload["upper"]},
            {"epoch": "bad"},
        ):
            signed = self.signed({**payload, **change}, "history-resume")
            response, body = await self.window(positive, {"resume": signed})
            self.assert_error(response, body, 400, "invalid-cursor")
        other = self._save_prompt("other positive")
        response, body = await self.window(other, {"resume": token})
        self.assert_error(response, body, 400, "invalid-cursor")
        with self.api.db.model.get_connection() as conn:
            conn.execute("UPDATE prompts SET text='changed' WHERE id=?", (positive,))
        response, body = await self.window(positive, {"resume": token})
        self.assert_error(response, body, 400, "invalid-cursor")
        source = str(self.comfy_dir / "resume-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, {"resume": token})
        self.assert_error(
            response, body, 409, "history-changed", {"action": "refresh-history"}
        )

    async def test_empty_history_has_no_authenticated_resume_anchor(self):
        positive = self._save_prompt("positive")
        response, body = await self.window(positive)
        self.assertEqual(response.status, 200, body)
        self.assertIn("resume_cursor", body)
        self.assertIsNone(body["resume_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)
        try:
            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"},
                )
        finally:
            writer.close_all()
        self.assertEqual(len(writer.model._connections), 0)

    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 = os.path.normcase(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)

    async def assert_projection_releases_retired_owner(
        self, route, *, acquisition=False, failure=None
    ):
        positive = self._save_prompt("positive")
        start = self.capture()
        await self.window(positive)
        old_db = self.api.db
        self.addCleanup(old_db.close_all)
        source = self.comfy_dir / "queued-restore.db"
        self.assertTrue(old_db.model.backup_database(str(source)))
        upload = source.read_bytes()
        reached = asyncio.Event()
        release = threading.Event() if acquisition else asyncio.Event()
        executor = self.api._run_in_executor
        connection = old_db.model.get_connection
        history = importlib.import_module("py.api.history")
        serialize = history._serialize_history
        fence = self.api._history_fence()
        validate = fence.validate
        loop = asyncio.get_running_loop()
        armed = True

        async def queued(func, *args, **kwargs):
            nonlocal armed
            if armed and func.__name__ == "project":
                armed = False
                reached.set()
                await asyncio.wait_for(release.wait(), HANDOFF_TIMEOUT_SECONDS)
            return await executor(func, *args, **kwargs)

        def selected_connection():
            nonlocal armed
            if armed:
                armed = False
                loop.call_soon_threadsafe(reached.set)
                if not release.wait(HANDOFF_TIMEOUT_SECONDS):
                    raise AssertionError("connection-acquisition barrier timed out")
            return connection()

        @contextmanager
        def failed_snapshot(expected_epoch=None):
            old_db.model.get_connection()
            raise sqlite3.OperationalError("snapshot unavailable")
            yield

        def released_before_serialization(payload, **kwargs):
            self.assertEqual(len(old_db.model._connections), 0)
            return serialize(payload, **kwargs)

        def released_before_validation(epoch, generation):
            self.assertEqual(len(old_db.model._connections), 0)
            return validate(epoch, generation)

        with ExitStack() as patches:
            patches.enter_context(
                patch.object(
                    history,
                    "_serialize_history",
                    side_effect=released_before_serialization,
                )
            )
            patches.enter_context(
                patch.object(fence, "validate", side_effect=released_before_validation)
            )
            if acquisition:
                patches.enter_context(
                    patch.object(
                        old_db.model, "get_connection", side_effect=selected_connection
                    )
                )
            else:
                patches.enter_context(
                    patch.object(self.api, "_run_in_executor", side_effect=queued)
                )
            if failure == "snapshot":
                patches.enter_context(
                    patch.object(
                        old_db, "recorded_history_snapshot", side_effect=failed_snapshot
                    )
                )
            elif failure == "lookup":
                patches.enter_context(
                    patch.object(
                        old_db,
                        (
                            "list_history_window"
                            if route == "window"
                            else "get_history_run"
                        ),
                        side_effect=sqlite3.OperationalError("lookup unavailable"),
                    )
                )
            read = (
                self.window(positive)
                if route == "window"
                else self.detail(start.run_id)
            )
            pending = asyncio.create_task(read)
            try:
                await asyncio.wait_for(reached.wait(), HANDOFF_TIMEOUT_SECONDS)
                form = FormData()
                form.add_field("database_file", upload, filename="upload.db")
                restored = await self.client.post("/prompt_manager/restore", data=form)
                self.assertEqual(restored.status, 200, await restored.text())
                self.assertIsNot(self.api.db, old_db)
                self.assertEqual(len(old_db.model._connections), 0)
            finally:
                release.set()
                response, body = await asyncio.wait_for(
                    pending, HANDOFF_TIMEOUT_SECONDS
                )

        if failure is not None:
            self.assert_error(
                response, body, 503, "storage-unavailable", {"action": "retry-read"}
            )
        elif acquisition:
            self.assert_error(
                response, body, 409, "history-changed", {"action": "refresh-history"}
            )
        else:
            self.assertEqual(response.status, 200, body)
            self.assertEqual(body["history_epoch"], self.api.db.get_execution_epoch())
            self.assertNotEqual(body["history_epoch"], start.storage_epoch)
            run = body["items"][0] if route == "window" else body["run"]
            self.assertEqual(run["run_id"], start.run_id)
            self.assertEqual(run["storage_epoch"], start.storage_epoch)
        self.assertEqual(
            len(old_db.model._connections),
            0,
            "completed projection retained a handle on the API-retired owner",
        )

    async def test_queued_window_across_multipart_restore_closes_retired_handle(self):
        await self.assert_projection_releases_retired_owner("window")

    async def test_queued_detail_across_multipart_restore_closes_retired_handle(self):
        await self.assert_projection_releases_retired_owner("detail")

    async def test_window_retirement_before_connection_acquisition_closes_handle(self):
        await self.assert_projection_releases_retired_owner("window", acquisition=True)

    async def test_detail_retirement_before_connection_acquisition_closes_handle(self):
        await self.assert_projection_releases_retired_owner("detail", acquisition=True)

    async def test_retired_window_handle_closes_after_snapshot_and_lookup_failures(
        self,
    ):
        for failure in ("snapshot", "lookup"):
            with self.subTest(failure=failure):
                await self.assert_projection_releases_retired_owner(
                    "window", acquisition=True, failure=failure
                )

    async def test_retired_detail_handle_closes_after_snapshot_and_lookup_failures(
        self,
    ):
        for failure in ("snapshot", "lookup"):
            with self.subTest(failure=failure):
                await self.assert_projection_releases_retired_owner(
                    "detail", acquisition=True, failure=failure
                )


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()