# Generated by python/build.mjs; do not hand-edit.
# ruff: noqa: E501
# MIT License
#
# Copyright (c) 2026 Naoya / jokv213
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
#
"""State Lens for an existing synchronous Playwright Page (MIT).
This source is a build template. Import the generated state_lens.py instead.
"""
from __future__ import annotations
import base64
import copy
import html
import inspect
import json
import math
import os
from contextlib import contextmanager
from datetime import UTC, datetime
from pathlib import Path
from urllib.parse import urlsplit
LIBRARY_VERSION = "0.7.0-experimental.1"
SOURCE_SHA256 = "6d1fb38045c6b01f3f072c135108d57ffa34b78256f28bc4822b512a0db4c797"
BROWSER_OBSERVER = "async ({ probe: p, timeout, expectedOrigin }) => {\n async function observeSelected() {\n if (location.origin !== expectedOrigin) return { status: 'INCONCLUSIVE', reason: 'Origin changed before reading the selected data' };\n function plain(value, undefinedPaths, datePaths, seen = new WeakSet(), depth = 0, path = '') {\n if (depth > 12) throw new Error('Selected value exceeds depth12');\n if (value === undefined) { undefinedPaths.push(path); return null; }\n if (value === null || ['string', 'boolean'].includes(typeof value)) return value;\n if (typeof value === 'number' && Number.isFinite(value)) return value;\n if (typeof value === 'object' && Object.prototype.toString.call(value) === '[object Date]') {\n if (!Number.isFinite(Date.prototype.getTime.call(value))) throw new Error('Selected Date value is invalid');\n datePaths.push(path);\n return Date.prototype.toISOString.call(value);\n }\n if (typeof value !== 'object' || Object.prototype.toString.call(value) !== '[object Object]' && !Array.isArray(value)) throw new Error('Selected value is not plain JSON');\n if (seen.has(value)) throw new Error('Selected value contains a cycle');\n seen.add(value);\n if (Array.isArray(value) && (Object.keys(value).length !== value.length || Array.from({ length: value.length }, (_, i) => !Object.hasOwn(value, i)).some(Boolean))) throw new Error('Sparse or extended arrays are not plain JSON');\n const childPath = key => `${path}/${String(key).replace(/~/g, '~0').replace(/\\//g, '~1')}`;\n const result = Array.isArray(value) ? value.map((x, i) => plain(x, undefinedPaths, datePaths, seen, depth + 1, childPath(i)))\n : Object.fromEntries(Object.entries(value).map(([k, v]) => [k, plain(v, undefinedPaths, datePaths, seen, depth + 1, childPath(k))]));\n seen.delete(value);\n return result;\n }\n function normalize(exists, value) {\n const pointed = p.pointer !== undefined;\n let selectedExists = exists;\n if (pointed && exists && p.pointer !== '') {\n for (const part of p.pointer.slice(1).split('/').map(x => x.replace(/~1/g, '/').replace(/~0/g, '~'))) {\n if (value === null || typeof value !== 'object' || !Object.hasOwn(value, part)\n || (Array.isArray(value) && (!/^(0|[1-9][0-9]*)$/.test(part) || !Number.isSafeInteger(Number(part)) || Number(part) >= value.length))) {\n selectedExists = false; break;\n }\n value = value[part];\n }\n }\n const undefinedPaths = [], datePaths = [];\n const normalized = selectedExists ? plain(value, undefinedPaths, datePaths) : null;\n undefinedPaths.sort(); datePaths.sort();\n const annotations = { ...(undefinedPaths.length ? { undefinedPaths } : {}), ...(datePaths.length ? { datePaths } : {}) };\n const selection = pointed ? { selection: { pointer: p.pointer, exists: selectedExists } } : {};\n const representation = pointed || undefinedPaths.length || datePaths.length ? { value: normalized, ...annotations, ...selection } : normalized;\n if (new TextEncoder().encode(JSON.stringify(representation)).byteLength > 65536) throw new Error('Selected value exceeds64KiB');\n return { status: 'OBSERVED', exists, value: normalized, ...selection,\n ...(datePaths.length ? { valueEncoding: 'json-with-type-paths-v1', ...annotations }\n : undefinedPaths.length ? { valueEncoding: 'json-with-undefined-paths-v1', undefinedPaths } : {}) };\n }\n try {\n if (p.source === 'localStorage' || p.source === 'sessionStorage') {\n const area = p.source === 'localStorage' ? localStorage : sessionStorage;\n const value = area.getItem(p.key);\n if (value === null || p.json !== true) return normalize(value !== null, value);\n if (p.pointer !== undefined && new TextEncoder().encode(value).byteLength > 4 * 1024 * 1024) throw new Error('Selected JSON text exceeds4MiB input limit');\n let parsed;\n try { parsed = JSON.parse(value); }\n catch (error) { if (p.pointer !== undefined) throw new Error('Selected JSON text is malformed'); throw error; }\n return normalize(true, parsed);\n }\n if (typeof indexedDB.databases !== 'function') throw new Error('Database existence cannot be checked');\n // Never enumerate or export app records; check that the selected database exists first.\n const databases = await Promise.race([\n indexedDB.databases(), new Promise((_, reject) => setTimeout(() => reject(new Error('Database listing timed out')), timeout)),\n ]);\n if (!databases.some(db => db.name === p.database)) throw new Error('Selected database does not exist');\n return await new Promise(resolve => {\n let db, tx, settled = false;\n const finish = result => {\n if (settled) return;\n settled = true; clearTimeout(timer);\n if (db) db.close();\n resolve(result);\n };\n const fail = message => finish({ status: 'INCONCLUSIVE', reason: message });\n const timer = setTimeout(() => { if (tx) { try { tx.abort(); } catch {} } fail('Selected database read timed out'); }, timeout);\n const request = indexedDB.open(p.database);\n request.onupgradeneeded = () => {\n // It disappeared between listing and open: abort creation, close the late result.\n request.transaction.abort();\n fail('Selected database disappeared before observation');\n };\n request.onerror = () => fail(request.error?.message || 'Database open failed');\n request.onblocked = () => fail('Selected database is blocked');\n request.onsuccess = () => {\n db = request.result;\n if (settled) { db.close(); return; }\n if (!db.objectStoreNames.contains(p.store)) { fail('Selected object store does not exist'); return; }\n try {\n tx = db.transaction(p.store, 'readonly');\n const store = tx.objectStore(p.store), key = store.getKey(p.key), value = store.get(p.key);\n tx.onerror = () => fail(tx.error?.message || 'Database transaction failed');\n tx.onabort = () => fail('Database transaction aborted');\n tx.oncomplete = () => { try { finish(normalize(key.result !== undefined, value.result)); } catch (error) { fail(error.message); } };\n } catch (error) { fail(error.message); }\n };\n });\n } catch (error) { return { status: 'INCONCLUSIVE', reason: error?.message || String(error) }; }\n }\n // Preserve JSON keys such as __proto__ across the browser protocol serializer.\n return JSON.stringify(await observeSelected());\n }"
HTML_HEAD = "State Lens · STATE_LENS_TITLE_6b138ce9\n\nSELECTED APPLICATION DATA · LOCAL OBSERVATIONS · State Lens 0.7.0-experimental.1
STATE_LENS_TITLE_6b138ce9
UI steps beside the data they leave behind.
Changes since the preceding checkpoint
Selected values
Snapshots are observations, not database commit timestamps, causal proof or assertions that all data is correct. INCONCLUSIVE is not absent or PASS. This file includes only the selected data and captured viewport, and stays local unless you share it.
\n"
TITLE_MARKER = "STATE_LENS_TITLE_6b138ce9"
_MISSING = object()
def _origin(url):
try:
u = urlsplit(url)
if u.scheme not in ("http", "https") or not u.hostname:
return None
host = u.hostname.lower()
host = f"[{host}]" if ":" in host else host.encode("idna").decode("ascii")
port = u.port
suffix = (
"" if port is None or (u.scheme, port) in (("http", 80), ("https", 443)) else f":{port}"
)
return f"{u.scheme}://{host}{suffix}"
except (ValueError, UnicodeError):
return None
def _validate(probes, timeout_ms):
if not isinstance(probes, list) or not 1 <= len(probes) <= 16:
raise ValueError("Select 1–16 application probes")
names = set()
for p in probes:
if (
not isinstance(p, dict)
or not isinstance(p.get("name"), str)
or not p["name"]
or p["name"] in names
):
raise ValueError("Probe names must be nonempty and unique")
names.add(p["name"])
if p.get("source") not in ("localStorage", "sessionStorage", "indexedDB"):
raise ValueError("Unsupported probe source")
if p["source"] in ("localStorage", "sessionStorage") and not isinstance(p.get("key"), str):
raise ValueError("Web Storage requires a string key")
if p["source"] == "indexedDB":
k = p.get("key")
if not all(isinstance(p.get(x), str) and p[x] for x in ("database", "store")) or not (
isinstance(k, str)
or type(k) in (int, float)
and math.isfinite(k)
and abs(k) <= 2**53 - 1
):
raise ValueError("IndexedDB requires database, store and a string/safe-number key")
if "pointer" in p:
pointer = p["pointer"]
if not isinstance(pointer, str) or pointer and not pointer.startswith("/"):
raise ValueError("pointer must be an RFC6901 JSON pointer")
if len(pointer.split("/")) > 33 or len(pointer.encode("utf-8")) > 2048:
raise ValueError("pointer exceeds32 segments or2048 UTF8 bytes")
for part in pointer.split("~")[1:]:
if not part or part[0] not in "01":
raise ValueError("Invalid RFC6901 escape")
if p["source"] != "indexedDB" and p.get("json") is not True:
raise ValueError("Web Storage pointer selection requires json:True")
if type(timeout_ms) is not int or not 50 <= timeout_ms <= 5000:
raise ValueError("timeout_ms must be50–5000")
json.dumps(probes, allow_nan=False)
def _encoded(value):
return json.dumps(value, ensure_ascii=True, allow_nan=False, separators=(",", ":"))
def differences(before, after, path="$", output=None):
out = [] if output is None else output
if before is not _MISSING and after is not _MISSING and _encoded(before) == _encoded(after):
return out
if isinstance(before, dict) and isinstance(after, dict):
for key in dict.fromkeys([*before, *after]):
differences(before.get(key, _MISSING), after.get(key, _MISSING), f"{path}.{key}", out)
elif isinstance(before, list) and isinstance(after, list):
for index in range(max(len(before), len(after))):
differences(
before[index] if index < len(before) else _MISSING,
after[index] if index < len(after) else _MISSING,
f"{path}.{index}",
out,
)
else:
out.append(
{
"path": path,
"before": "(missing)" if before is _MISSING else before,
"after": "(missing)" if after is _MISSING else after,
}
)
return out
def render_timeline(report):
data = (
_encoded(report)
.replace("<", "\\u003c")
.replace("\u2028", "\\u2028")
.replace("\u2029", "\\u2029")
)
head = HTML_HEAD.replace(TITLE_MARKER, html.escape(str(report["title"]), quote=True))
return head + data + HTML_TAIL
class StateLens:
"""Observe selected state on one sync Page; keep your own waits/assertions."""
def __init__(
self,
page,
*,
probes,
timeout_ms=750,
screenshots=True,
title="What happened to the saved data?",
):
_validate(probes, timeout_ms)
if inspect.iscoroutinefunction(page.evaluate) or not isinstance(page.url, str):
raise TypeError("Use a synchronous Playwright Page; async API is not supported")
self._page = page
self._probes = copy.deepcopy(probes)
self._timeout = timeout_ms
self._screenshots = screenshots
self._title = title
self._checkpoints = []
self._running = False
self._selected_origin = _origin(page.url)
def _capture(self, label, action):
url = self._page.url
origin = _origin(url)
if self._selected_origin is None and origin is not None:
self._selected_origin = origin
in_scope = origin is not None and origin == self._selected_origin
values = {}
for probe in self._probes:
try:
values[probe["name"]] = (
json.loads(
self._page.evaluate(
BROWSER_OBSERVER,
{
"probe": probe,
"timeout": self._timeout,
"expectedOrigin": self._selected_origin,
},
)
)
if in_scope
else {
"status": "INCONCLUSIVE",
"reason": "Outside the first selected HTTP origin",
}
)
except Exception as error:
values[probe["name"]] = {"status": "INCONCLUSIVE", "reason": str(error)}
screenshot = screenshot_error = None
if self._screenshots and in_scope and _origin(self._page.url) == origin:
try:
screenshot = base64.b64encode(
self._page.screenshot(type="png", timeout=3000)
).decode("ascii")
except Exception as error:
screenshot_error = str(error)
if _origin(self._page.url) != origin:
in_scope = False
screenshot = None
values = {
p["name"]: {
"status": "INCONCLUSIVE",
"reason": "Origin changed during the sequential observation",
}
for p in self._probes
}
previous = self._checkpoints[-1] if self._checkpoints else None
same_scope = (
in_scope
and previous
and previous["origin"] == origin
and previous["scopeStatus"] == "SELECTED_ORIGIN"
)
checkpoint = {
"index": len(self._checkpoints),
"label": label,
"observedAt": datetime.now(UTC)
.isoformat(timespec="milliseconds")
.replace("+00:00", "Z"),
"url": url,
"endURL": self._page.url,
"origin": origin,
"selectedOrigin": self._selected_origin,
"scopeStatus": "SELECTED_ORIGIN" if in_scope else "OUTSIDE_ORIGIN",
"action": action,
"values": values,
"changes": differences(previous["values"], values) if same_scope else [],
"screenshot": screenshot,
"screenshotError": screenshot_error,
}
self._checkpoints.append(checkpoint)
return copy.deepcopy(checkpoint)
def checkpoint(self, label):
if self._running:
raise RuntimeError("Overlapping steps are not supported")
self._running = True
try:
return self._capture(label, {"status": "CHECKPOINT"})
finally:
self._running = False
@contextmanager
def step(self, label):
if self._running:
raise RuntimeError("Overlapping steps are not supported")
self._running = True
try:
self._capture(f"{label} · before", {"status": "BEFORE"})
try:
yield
except Exception as error:
self._capture(f"{label} · failed", {"status": "FAILED", "error": str(error)})
raise
else:
self._capture(f"{label} · after", {"status": "COMPLETED"})
finally:
self._running = False
def report(self):
return copy.deepcopy(
{
"schema": 4,
"libraryVersion": LIBRARY_VERSION,
"title": self._title,
"probes": self._probes,
"checkpoints": self._checkpoints,
}
)
def write_html(self, path):
"""Write a new local report; never replace a previous report."""
text = render_timeline(self.report())
fd = os.open(Path(path), os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
with os.fdopen(fd, "w", encoding="utf-8") as stream:
stream.write(text)