diff --git a/__init__.py b/__init__.py index 2663b6aa..f4945892 100644 --- a/__init__.py +++ b/__init__.py @@ -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 diff --git a/py/api/__init__.py b/py/api/__init__.py index c085e63a..404e0c4f 100644 --- a/py/api/__init__.py +++ b/py/api/__init__.py @@ -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) diff --git a/py/api/history.py b/py/api/history.py new file mode 100644 index 00000000..53a60d0b --- /dev/null +++ b/py/api/history.py @@ -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)) diff --git a/tests/js/api-client.test.js b/tests/js/api-client.test.js index 55951117..211f0643 100644 --- a/tests/js/api-client.test.js +++ b/tests/js/api-client.test.js @@ -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; + }); + } + } +}); diff --git a/tests/test_api_history.py b/tests/test_api_history.py new file mode 100644 index 00000000..2e8c77af --- /dev/null +++ b/tests/test_api_history.py @@ -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() diff --git a/tests/test_history_read_fence.py b/tests/test_history_read_fence.py new file mode 100644 index 00000000..49eecf6f --- /dev/null +++ b/tests/test_history_read_fence.py @@ -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) diff --git a/utils/history_read_fence.py b/utils/history_read_fence.py new file mode 100644 index 00000000..dd78c61b --- /dev/null +++ b/utils/history_read_fence.py @@ -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() diff --git a/web/js/api-client.js b/web/js/api-client.js index e4d4aa49..23d65fa8 100644 --- a/web/js/api-client.js +++ b/web/js/api-client.js @@ -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; }