Vito
ecd21165a7
* feat(history): add bounded chronological read projections (#209) * fix(history): validate internal continuation positions (#209) * feat(history): expose fenced history windows and run details (#209) * fix(history): release retired projection connections (#209) * feat(history): resolve reported outputs through current authorization (#209) * fix(history): reject invalid inspection targets (#209) * feat(history): add scoped history state and current-file adapter (#209) * fix(history): preserve slideshow and clear stale reentry actions (#209) * feat(history): render recorded executions and frozen run details (#209) * fix(history): recover receipts and refresh capture evidence (#209) * test(history): verify chronological history across restore and host modes (#209) * fix(history): enforce chronological marker visual parity (#209) * fix(history): resolve final chronological review findings Address #209 recovery, bounded visit restoration, exact receipt availability, and portable regression coverage. * test(history): repair Windows fixtures and review feedback Address #209 and PR #227 native path, cleanup, media and review-comment expectations. Document Windows portability preflight without changing production behavior. * ci: raise the viewerjs job budget for the larger browser suite The cross-browser workflows suite grew from 396 to 496 tests on this branch, which pushed the whole viewerjs job to about 25.5 minutes against the 25 minute cap set in #181. Two exact-head runs on this branch timed out without any test failing: run 39173019283 was killed on test 474 of 496, and run 39175125474 printed "496 passed (12.8m)" and was then cancelled during Playwright shutdown. Raise only the job budget. Browser projects, workers and per-test timeouts are unchanged. Splitting the browsers into parallel jobs is tracked separately.
539 lines
19 KiB
Python
539 lines
19 KiB
Python
"""Real disposable ledger commands and physical files for history workflows."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import replace
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import threading
|
|
from types import SimpleNamespace
|
|
from typing import TYPE_CHECKING
|
|
from uuid import UUID, uuid5
|
|
|
|
from aiohttp import web
|
|
from PIL import Image
|
|
from PIL.PngImagePlugin import PngInfo
|
|
|
|
from tests.browser.workspace_visual_fixture import _write_art, seed_workspace_visual
|
|
from utils.execution_contexts import (
|
|
ContextBundle,
|
|
PromptSnapshot,
|
|
RecipeContext,
|
|
TextEvidence,
|
|
)
|
|
from utils.execution_history_events import make_finish, make_start
|
|
from utils.execution_outputs import OutputBundle, OutputReference, snapshot_output_roots
|
|
|
|
if TYPE_CHECKING:
|
|
from py.api import PromptManagerAPI
|
|
|
|
|
|
HISTORY_POSITIVE = (
|
|
"Chronology A,A,B\nExact positive 猫 😀 " + "fine light " * 35
|
|
).rstrip()
|
|
DETAIL_POSITIVE = "Frozen evidence\n" + "unbroken-" * 90
|
|
LONG_NEGATIVE = "same visible negative\n" + "exact detail 猫 😀 " * 70
|
|
SESSION = "69f04366-6384-4182-b84b-117597479127"
|
|
|
|
|
|
class _HistoryCommands:
|
|
"""One disposable fixture lifetime and one bounded publication barrier."""
|
|
|
|
def __init__(self, api: PromptManagerAPI, root: Path) -> None:
|
|
self.api = api
|
|
self.root = root
|
|
self.manifest = None
|
|
self.entered = threading.Event()
|
|
self.release = threading.Event()
|
|
self.restore_seam = None
|
|
import folder_paths
|
|
|
|
self.original_output = folder_paths.get_output_directory
|
|
|
|
def reset(self, manifest: dict | None = None) -> None:
|
|
self.disarm()
|
|
import folder_paths
|
|
|
|
folder_paths.get_output_directory = self.original_output
|
|
self.manifest = manifest
|
|
|
|
def disarm(self) -> None:
|
|
self.release.set()
|
|
if self.restore_seam:
|
|
module, name, original = self.restore_seam
|
|
setattr(module, name, original)
|
|
self.restore_seam = None
|
|
|
|
def arm(self, boundary: str) -> None:
|
|
from py.api import history
|
|
from py.api import inspection_service
|
|
|
|
self.disarm()
|
|
self.entered = threading.Event()
|
|
self.release = threading.Event()
|
|
name = {
|
|
"window": "_serialize_history",
|
|
"detail": "_serialize_history",
|
|
"candidate": "resolve_current_candidate",
|
|
"inspection": "read_inspection_file",
|
|
"media": "_prepare_current_media",
|
|
}[boundary]
|
|
module = inspection_service if boundary == "inspection" else history
|
|
original = getattr(module, name)
|
|
entered, release = self.entered, self.release
|
|
|
|
def held(*args: object, **kwargs: object) -> object:
|
|
value = original(*args, **kwargs)
|
|
setattr(module, name, original)
|
|
entered.set()
|
|
try:
|
|
if not release.wait(10):
|
|
raise TimeoutError("Owned history publication barrier timed out")
|
|
return value
|
|
except BaseException:
|
|
if boundary == "media":
|
|
value.close()
|
|
raise
|
|
|
|
self.restore_seam = module, name, original
|
|
setattr(module, name, held)
|
|
|
|
|
|
def register_workspace_history_commands(
|
|
routes: web.RouteTableDef, api: PromptManagerAPI, root: Path
|
|
) -> _HistoryCommands:
|
|
"""Register fixed commands; all files and namespaces belong to this server."""
|
|
state = _HistoryCommands(api, root)
|
|
api._workspace_history_commands = state
|
|
|
|
@routes.post("/__test__/history/commands")
|
|
async def command(request: web.Request) -> web.Response:
|
|
data = await request.json() if request.can_read_body else None
|
|
actions = {
|
|
"release",
|
|
"replace-file",
|
|
"reconfigure-root",
|
|
"make-missing-available",
|
|
"append-history-runs",
|
|
}
|
|
if (
|
|
not isinstance(data, dict)
|
|
or type(data.get("action")) is not str
|
|
or ("boundary" in data and type(data["boundary"]) is not str)
|
|
or not (
|
|
(set(data) == {"action"} and data["action"] in actions)
|
|
or (
|
|
set(data) == {"action", "boundary"}
|
|
and data["action"] == "arm"
|
|
and data["boundary"]
|
|
in {"window", "detail", "candidate", "inspection", "media"}
|
|
)
|
|
)
|
|
):
|
|
raise web.HTTPBadRequest()
|
|
if state.manifest is None:
|
|
raise web.HTTPConflict()
|
|
action = data["action"]
|
|
if action == "arm":
|
|
state.arm(data["boundary"])
|
|
elif action == "release":
|
|
state.disarm()
|
|
elif action == "replace-file":
|
|
replacement = root / "output" / "history" / "replacement.png"
|
|
with Image.new("RGB", (160, 100), (225, 30, 40)) as image:
|
|
image.save(replacement)
|
|
replacement.replace(root / "output" / "history" / "current.png")
|
|
elif action == "make-missing-available":
|
|
target = root / "output" / "history" / "missing.png"
|
|
with Image.new("RGB", (160, 100), (30, 150, 70)) as image:
|
|
image.save(target)
|
|
elif action == "append-history-runs":
|
|
session = str(uuid5(UUID(SESSION), "fixed-late-history"))
|
|
context = RecipeContext(
|
|
"10", _snapshot(HISTORY_POSITIVE), _snapshot("A negative"), "inferred"
|
|
)
|
|
for number in range(61):
|
|
job = str(uuid5(UUID(session), str(number)))
|
|
start = make_start(
|
|
session,
|
|
((0, job, {}, {}), number),
|
|
api.db.get_execution_epoch(),
|
|
1_790_000_000_000 + number,
|
|
)
|
|
api.db.write_execution_start(start)
|
|
api.db.write_execution_finish(
|
|
make_finish(
|
|
session,
|
|
number,
|
|
start.storage_epoch,
|
|
SimpleNamespace(
|
|
messages=[("execution_success", {"prompt_id": job})]
|
|
),
|
|
1_790_000_000_010 + number,
|
|
)
|
|
)
|
|
api.db.write_execution_contexts(
|
|
start.run_id,
|
|
start.storage_epoch,
|
|
ContextBundle((context,), "complete", "captured"),
|
|
)
|
|
elif action == "reconfigure-root":
|
|
import folder_paths
|
|
|
|
moved = root / "owned-reconfigured-output"
|
|
moved.mkdir(exist_ok=True)
|
|
folder_paths.get_output_directory = lambda: str(moved)
|
|
return web.json_response({"success": True})
|
|
|
|
@routes.get("/__test__/history/checkpoint")
|
|
async def checkpoint(request: web.Request) -> web.Response:
|
|
if state.manifest is None:
|
|
raise web.HTTPConflict()
|
|
return web.json_response({"entered": state.entered.is_set()})
|
|
|
|
@routes.get("/__test__/history/snapshot")
|
|
async def snapshot(request: web.Request) -> web.Response:
|
|
if state.manifest is None or request.query:
|
|
raise web.HTTPConflict()
|
|
path = root / "owned-history-snapshot.db"
|
|
created = await api._run_in_executor(api.db.model.backup_database, str(path))
|
|
if not created:
|
|
raise web.HTTPInternalServerError()
|
|
return web.Response(
|
|
body=path.read_bytes(), content_type="application/octet-stream"
|
|
)
|
|
|
|
@routes.get("/__test__/history/reused-snapshot")
|
|
async def reused_snapshot(request: web.Request) -> web.Response:
|
|
if state.manifest is None or request.query:
|
|
raise web.HTTPConflict()
|
|
from database.operations import PromptDatabase
|
|
from utils.hashing import generate_prompt_hash
|
|
|
|
path = root / "owned-history-reused.db"
|
|
|
|
def create() -> bytes:
|
|
for suffix in ("", "-wal", "-shm"):
|
|
Path(str(path) + suffix).unlink(missing_ok=True)
|
|
database = PromptDatabase(str(path))
|
|
try:
|
|
target = state.manifest["detail_id"]
|
|
with database.model.get_connection() as connection:
|
|
text = "Restored owner with reused numeric ID"
|
|
connection.execute(
|
|
"INSERT INTO prompts(id,text,hash) VALUES(?,?,?)",
|
|
(target, text, generate_prompt_hash(text)),
|
|
)
|
|
database.save_prompt_pair(text, "Restored saved negative")
|
|
finally:
|
|
database.close_all()
|
|
return path.read_bytes()
|
|
|
|
return web.Response(
|
|
body=await api._run_in_executor(create),
|
|
content_type="application/octet-stream",
|
|
)
|
|
|
|
async def cleanup(application: web.Application) -> None:
|
|
state.reset()
|
|
|
|
api_app = __import__("server").PromptServer.instance.app
|
|
api_app.on_cleanup.append(cleanup)
|
|
return state
|
|
|
|
|
|
def _known(value: str, source: str = "runtime_return") -> TextEvidence:
|
|
provenance = "inferred" if source.startswith("graph_") else "recorded"
|
|
return TextEvidence(value, provenance, source, None)
|
|
|
|
|
|
def _snapshot(raw: str, suffix: str = "") -> PromptSnapshot:
|
|
return PromptSnapshot(_known(raw), _known(raw + suffix), _known(""), _known(suffix))
|
|
|
|
|
|
def capture_workspace_history_zero_outputs(api: PromptManagerAPI, run_id: str) -> bool:
|
|
"""Apply one late host capture through the normal immutable storage command."""
|
|
run = api.db.get_execution_run(run_id)
|
|
finish = make_finish(
|
|
run["session_id"],
|
|
run["queue_item_id"],
|
|
run["storage_epoch"],
|
|
SimpleNamespace(
|
|
messages=[
|
|
(
|
|
"execution_success",
|
|
{"prompt_id": run["job_id"], "timestamp": run["terminal_event_at"]},
|
|
)
|
|
]
|
|
),
|
|
run["observed_callback_at"],
|
|
)
|
|
return api.db.write_execution_outputs_for_finish(
|
|
finish, OutputBundle((), "complete", "none_reported")
|
|
)
|
|
|
|
|
|
def seed_workspace_history(
|
|
api: PromptManagerAPI, root: Path, *, history_runs: int = 0
|
|
) -> dict:
|
|
"""Write actual captures; reset removes only this server's disposable rows."""
|
|
api._workspace_history_commands.reset()
|
|
seed_workspace_visual(api, root)
|
|
with api.db.model.get_connection() as conn:
|
|
for table in (
|
|
"execution_run_outputs",
|
|
"execution_output_capture",
|
|
"execution_run_contexts",
|
|
"execution_context_capture",
|
|
"execution_runs",
|
|
"execution_contexts",
|
|
"execution_prompt_snapshots",
|
|
"execution_context_texts",
|
|
):
|
|
conn.execute(f"DELETE FROM {table}")
|
|
epoch = api.db.get_execution_epoch()
|
|
roots = snapshot_output_roots("posix", str(root / "output"), str(root / "temp"))
|
|
folder = root / "output" / "history"
|
|
folder.mkdir(exist_ok=True)
|
|
for index, name in enumerate(("current.png", "other.png")):
|
|
with Image.new("RGB", (160, 100), (50, 80 + index * 50, 110)) as image:
|
|
image.save(folder / name)
|
|
|
|
def reference(
|
|
node: str, ordinal: int, name: str, kind: str = "output"
|
|
) -> OutputReference:
|
|
return OutputReference(
|
|
node,
|
|
ordinal,
|
|
kind,
|
|
"history",
|
|
name,
|
|
"posix",
|
|
roots.output_key if kind == "output" else roots.temp_key,
|
|
)
|
|
|
|
item = 0
|
|
|
|
def run(
|
|
contexts: tuple[RecipeContext, ...],
|
|
outputs: OutputBundle | None = None,
|
|
*,
|
|
observed: int | None = None,
|
|
conflicting: bool = False,
|
|
) -> str:
|
|
nonlocal item
|
|
job = str(uuid5(UUID(SESSION), f"job-{item}"))
|
|
start = make_start(
|
|
SESSION,
|
|
((item, job, {}, {}), item),
|
|
epoch,
|
|
1_780_000_000_000 + item if observed is None else observed,
|
|
)
|
|
api.db.write_execution_start(start)
|
|
messages = [("execution_success", {"prompt_id": job, "timestamp": item + 1})]
|
|
finish = make_finish(
|
|
SESSION, item, epoch, SimpleNamespace(messages=messages), item + 10
|
|
)
|
|
api.db.write_execution_finish(finish)
|
|
partial = any(
|
|
getattr(side, field).value is None
|
|
for context in contexts
|
|
for side in (context.positive, context.negative)
|
|
if side is not None
|
|
for field in ("raw", "effective", "prepend", "append")
|
|
)
|
|
api.db.write_execution_contexts(
|
|
start.run_id,
|
|
epoch,
|
|
ContextBundle(
|
|
contexts,
|
|
"partial" if partial else "complete",
|
|
"partial" if partial else "captured",
|
|
),
|
|
)
|
|
if outputs is not None:
|
|
api.db.write_execution_outputs_for_finish(finish, outputs)
|
|
if conflicting:
|
|
api.db.write_execution_contexts(
|
|
start.run_id,
|
|
epoch,
|
|
ContextBundle(
|
|
(replace(contexts[0], negative=_snapshot("later")),),
|
|
"complete",
|
|
"captured",
|
|
),
|
|
)
|
|
api.db.write_execution_outputs_for_finish(
|
|
finish, OutputBundle((), "complete", "none_reported")
|
|
)
|
|
item += 1
|
|
return start.run_id
|
|
|
|
pair_a = api.db.save_prompt_pair(
|
|
HISTORY_POSITIVE, "A negative", title="Saved A", category="History"
|
|
)
|
|
api.db.save_prompt_pair(HISTORY_POSITIVE, "B negative", title="Saved B")
|
|
positive_id = api.db.get_prompt_pair(pair_a)["positive_prompt_id"]
|
|
image = api.db.link_image_to_prompt(positive_id, str(folder / "current.png"))
|
|
api.db.link_image_to_pair(pair_a, image, {"source": "history browser fixture"})
|
|
a = RecipeContext(
|
|
"10", _snapshot(HISTORY_POSITIVE), _snapshot("A negative"), "inferred"
|
|
)
|
|
b = replace(a, negative=_snapshot("B negative"))
|
|
aab = [run((context,)) for context in (a, a, b)]
|
|
for number in range(history_runs):
|
|
context = (
|
|
a
|
|
if number % 3 == 0
|
|
else (
|
|
b
|
|
if number % 3 == 1
|
|
else replace(b, negative=_snapshot(f"Distinct context {number:04}"))
|
|
)
|
|
)
|
|
run((context,))
|
|
|
|
saved_pair = api.db.save_prompt_pair(
|
|
DETAIL_POSITIVE, LONG_NEGATIVE, title="Saved exact negative"
|
|
)
|
|
detail_id = api.db.get_prompt_pair(saved_pair)["positive_prompt_id"]
|
|
api.db.save_prompt_pair(DETAIL_POSITIVE, "", title="Saved empty")
|
|
current = api.db.link_image_to_prompt(detail_id, str(folder / "current.png"))
|
|
api.db.link_image_to_pair(saved_pair, current, {"source": "fixture"})
|
|
x = RecipeContext(
|
|
"10",
|
|
_snapshot(DETAIL_POSITIVE, " + composition A"),
|
|
_snapshot(LONG_NEGATIVE, " A"),
|
|
"inferred",
|
|
)
|
|
y = replace(x, negative=_snapshot(LONG_NEGATIVE, " B"))
|
|
receipts = OutputBundle(
|
|
(
|
|
reference("10", 0, "current.png"),
|
|
reference("10", 1, "current.png"),
|
|
reference("2", 0, "other.png"),
|
|
reference("2", 1, "missing.png"),
|
|
reference("z", 0, "transient.png", "temp"),
|
|
),
|
|
"complete",
|
|
"reported",
|
|
)
|
|
detail_runs = {
|
|
"shared": run(
|
|
(x, replace(x, anchor_id="2"), replace(y, anchor_id="z")), receipts
|
|
),
|
|
"equal_looking": run((y,)),
|
|
"empty": run(
|
|
(replace(x, negative=_snapshot("")),),
|
|
OutputBundle((), "complete", "none_reported"),
|
|
),
|
|
"unknown": run(
|
|
(
|
|
replace(
|
|
x,
|
|
negative=PromptSnapshot(
|
|
TextEvidence(None, "unknown", "unknown", "missing"),
|
|
_known(" \n\t ", "host_cache_output"),
|
|
_known("", "graph_literal"),
|
|
_known("", "graph_semantics"),
|
|
),
|
|
),
|
|
)
|
|
),
|
|
"whitespace": run(
|
|
(
|
|
replace(
|
|
x,
|
|
negative=PromptSnapshot(
|
|
_known(" \n\t ", "graph_literal"),
|
|
_known(" \n\t ", "host_cache_output"),
|
|
_known("", "graph_literal"),
|
|
_known("", "graph_semantics"),
|
|
),
|
|
),
|
|
),
|
|
OutputBundle(
|
|
(reference("1", 0, "current.png"),), "partial", "limit_reached"
|
|
),
|
|
),
|
|
"conflicting": run(
|
|
(x,), receipts, conflicting=True, observed=9_223_372_036_854_775_807
|
|
),
|
|
"paged": run(
|
|
(x,),
|
|
OutputBundle(
|
|
tuple(
|
|
reference("paged", ordinal, "current.png") for ordinal in range(41)
|
|
),
|
|
"complete",
|
|
"reported",
|
|
),
|
|
),
|
|
}
|
|
canonical_positive = (
|
|
"Layered sandstone arches, warm afternoon light, precise silhouettes."
|
|
)
|
|
canonical_negative = "blurred edges, watermark"
|
|
canonical_pair = api.db.save_prompt_pair(
|
|
canonical_positive, canonical_negative, title="Quality baseline"
|
|
)
|
|
canonical_id = api.db.get_prompt_pair(canonical_pair)["positive_prompt_id"]
|
|
api.db.save_prompt_pair(canonical_positive, "flat light", title="Clean geometry")
|
|
api.db.save_prompt_pair(canonical_positive, "", title="No negative")
|
|
for number, name in enumerate(("canonical.png", "canonical-alt.png")):
|
|
path = folder / name
|
|
_write_art(path, number, canonical_positive, canonical_negative)
|
|
with Image.open(path) as source:
|
|
graph = json.loads(source.info["prompt"])
|
|
graph["3"]["inputs"].update(
|
|
seed=42001 + number,
|
|
steps=28,
|
|
cfg=6.5,
|
|
sampler_name="euler",
|
|
scheduler="normal",
|
|
)
|
|
metadata = PngInfo()
|
|
metadata.add_text("prompt", json.dumps(graph))
|
|
source.save(path, pnginfo=metadata)
|
|
stamp = 1_760_000_000 - number * 60
|
|
os.utime(path, (stamp, stamp))
|
|
image_id = api.db.link_image_to_prompt(canonical_id, str(path))
|
|
api.db.link_image_to_pair(canonical_pair, image_id, {"source": "fixture"})
|
|
canonical_run = run(
|
|
(
|
|
RecipeContext(
|
|
"10",
|
|
_snapshot(canonical_positive, " + frozen composition"),
|
|
_snapshot(canonical_negative),
|
|
"inferred",
|
|
),
|
|
),
|
|
OutputBundle(
|
|
(
|
|
reference("10", 0, "canonical.png"),
|
|
reference("10", 1, "canonical-alt.png"),
|
|
),
|
|
"complete",
|
|
"reported",
|
|
),
|
|
)
|
|
merge_duplicate_id = api.db.save_prompt(DETAIL_POSITIVE)
|
|
api.invalidate_gallery_cache()
|
|
manifest = {
|
|
"positive_id": positive_id,
|
|
"detail_id": detail_id,
|
|
"aab": aab,
|
|
"runs": detail_runs,
|
|
"saved_pair": saved_pair,
|
|
"detail_image_id": current,
|
|
"canonical_id": canonical_id,
|
|
"canonical_pair": canonical_pair,
|
|
"canonical_run": canonical_run,
|
|
"canonical_positive": canonical_positive,
|
|
"canonical_negative": canonical_negative,
|
|
"merge_duplicate_id": merge_duplicate_id,
|
|
}
|
|
api._workspace_history_commands.reset(manifest)
|
|
return manifest
|