File size: 17,704 Bytes
bc50ea0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 | """Scoped, preview-confirmed deletion of processed takes (never originals)."""
import hashlib
import json
import os
from pathlib import Path, PurePosixPath
import re
import uuid
from .checkpoint_manager import (
CheckpointDeleteBlocked, _strict_run_name, checkpoint_run_lock,
)
from .checkpoint_variants import validate_processing_lineage
REVISION = re.compile(r"clip_(\d{4})\.([0-9a-f]{32})\.json")
POINTER = re.compile(r"clip_\d{4}\.json")
MANIFESTS = {"h3_chain_upscale_manifest_v1", "h3_chain_upscale_partial_manifest_v1"}
ARTIFACTS = {
"segment": ("segments", ".mp4", "Processed video"),
"checkpoint": ("checkpoints", ".safetensors", "Processed checkpoint"),
"prompt_file": ("prompts", ".txt", "Prompt snapshot"),
"generated_audio": ("audio", ".wav", "Processed audio"),
}
def _independent_pixel_take(metadata):
"""Only the saved pixel backend proves that HQ prefixes were not used.
context_steps alone describes the outgoing tail, not incoming context.
Never infer independence from a profile name, a UI stage, or missing data.
"""
config = metadata.get("profile_config")
segment = metadata.get("segment")
return (isinstance(config, dict) and config.get("backend") == "pixel"
and isinstance(segment, dict)
and type(segment.get("context_steps")) is int
and segment["context_steps"] == 0)
class ProcessingCheckpointManager:
def __init__(self, output_root):
self.root = Path(output_root).resolve()
def _path(self, address):
if not isinstance(address, str) or not address:
raise ValueError("A saved processing artifact address is required.")
parts = PurePosixPath(address).parts
if (PurePosixPath(address).is_absolute() or ".." in parts or
"\\" in address or str(PurePosixPath(address)) != address):
raise ValueError("Invalid processing artifact address.")
path = self.root
for part in parts:
path /= part
if path.is_symlink():
raise ValueError("Processing deletion cannot follow symlinks: %s" % address)
if not path.resolve().is_relative_to(self.root):
raise ValueError("Processing artifact escapes the output directory.")
if path.exists() and not (path.is_file() or path.is_dir()):
raise ValueError("Unsupported processing artifact: %s" % address)
return path
def _address(self, path):
return path.relative_to(self.root).as_posix()
def _read(self, path):
self._path(self._address(path))
value = json.loads(path.read_text(encoding="utf-8"))
if not isinstance(value, dict):
raise ValueError("Processing metadata must be an object: %s" % path.name)
return value
def _target(self, run, address):
path = self._path(address)
parts = PurePosixPath(address).parts
if not (parts[:2] == ("h3_chains", run) and (
len(parts) == 6 and parts[2] == "upscaled" or
len(parts) == 8 and parts[2] == "chapters" and parts[4] == "upscaled")
and parts[-2] == "checkpoints" and REVISION.fullmatch(parts[-1])):
raise ValueError("Select an immutable processed checkpoint inside this run.")
if not path.is_file():
raise FileNotFoundError("The selected processed take no longer exists.")
return path, path.parent.parent
def _documents(self, run):
run_dir = self._path("h3_chains/" + run)
parents = [run_dir / "upscaled"]
chapters = self._path(self._address(run_dir / "chapters"))
if chapters.is_dir():
for chapter in chapters.iterdir():
self._path(self._address(chapter))
if chapter.is_dir():
parents.append(chapter / "upscaled")
docs = {}
for parent in parents:
self._path(self._address(parent))
if not parent.is_dir():
continue
for profile in sorted(parent.iterdir()):
self._path(self._address(profile))
if not profile.is_dir():
continue
folder = self._path(self._address(profile / "checkpoints"))
files = list(folder.glob("clip_*.json")) if folder.is_dir() else []
files.append(profile / "upscale_manifest.json")
partial = self._path(self._address(profile / "partial"))
if partial.is_dir():
files.extend(partial.glob("through_clip_*.manifest.json"))
for path in sorted(files):
self._path(self._address(path))
if not path.exists():
continue
value = self._read(path)
if (value.get("run_name") != run or value.get("profile") != profile.name or
value.get("format") not in MANIFESTS | {"h3_chain_upscale_segment_v1"}):
raise ValueError("Cannot verify processing metadata: %s" % self._address(path))
if path.parent == folder:
if value["format"] != "h3_chain_upscale_segment_v1":
raise ValueError("Not a processing checkpoint: %s" % path.name)
match = REVISION.fullmatch(path.name)
if not match and not POINTER.fullmatch(path.name):
raise ValueError("Unrecognized processing checkpoint: %s" % path.name)
segment = value.get("segment")
if not isinstance(segment, dict):
raise ValueError("Processing checkpoint has no segment: %s" % path.name)
scene, revision = segment.get("index"), segment.get("revision")
if (type(scene) is not int or scene < 1 or
not re.fullmatch(r"[0-9a-f]{32}", str(revision)) or
path.name != ("clip_%04d.%s.json" % (scene, revision) if match
else "clip_%04d.json" % scene)):
raise ValueError("Processing revision identity mismatch: %s" % path.name)
canonical = folder / ("clip_%04d.%s.json" % (scene, revision))
if self._path(segment.get("revision_metadata")) != canonical:
raise ValueError("Processing revision address mismatch: %s" % path.name)
if "processing_lineage" in value:
validate_processing_lineage(value["processing_lineage"])
elif not isinstance(value.get("segments"), list):
raise ValueError("Processing manifest has no segment list: %s" % path.name)
docs[path] = value
return docs
def _preview(self, run, address):
target, profile = self._target(run, address)
docs = self._documents(run)
metadata = docs[target]
segment = metadata["segment"]
scene, revision = segment["index"], segment["revision"]
stem = "clip_%04d.%s" % (scene, revision)
owned = {target: "Revision metadata"}
for field, (folder, suffix, label) in ARTIFACTS.items():
if not segment.get(field):
continue # Missing optional or already-lost artifacts do not prevent cleanup.
path = self._path(segment[field])
if path != profile / folder / (stem + suffix):
raise ValueError("Processed take does not own its %s path." % field)
owned[path] = label
pointer = profile / "checkpoints" / ("clip_%04d.json" % scene)
if pointer in docs and docs[pointer]["segment"]["revision"] == revision:
owned[pointer] = "Current processed-take pointer (cleared, not rolled back)"
def refers(value):
if isinstance(value, dict):
if any(value.get(k) == revision for k in (
"predecessor_revision", "previous_revision", "context_revision")):
return True
if (value.get("source_revision") == revision and
(not value.get("source_checkpoint") or value["source_checkpoint"] in addresses)):
return True
address = value.get("revision_metadata", value.get("metadata_path"))
if (value.get("revision") == revision and value.get("index", value.get("scene")) == scene
and (not address or address in addresses)):
return True
return any(refers(v) for k, v in value.items() if k != "supersedes")
if isinstance(value, list):
return any(refers(v) for v in value)
return isinstance(value, str) and value in addresses
addresses = {self._address(path) for path in owned if path != pointer}
independent = _independent_pixel_take(metadata)
retained = {}
def independent_successor(item):
if not independent or not isinstance(item, dict) or not item.get("revision_metadata"):
return False
path = self._path(item["revision_metadata"])
saved = docs.get(path)
if (path.parent.parent != profile or not saved or not _independent_pixel_take(saved)
or saved["segment"]["index"] <= scene
or any(saved["segment"].get(k) != item.get(k)
for k in ("index", "revision", "checkpoint_sha256"))):
return False
retained[item["revision_metadata"]] = {
"scene": item["index"], "revision": item["revision"],
"metadata_path": item["revision_metadata"],
}
return True
dependents = []
for path, value in docs.items():
if path in owned:
continue
if value["format"] in MANIFESTS:
if refers(value.get("source_manifest")):
dependents.append({"metadata_path": self._address(path),
"reason": "uses this take as a processing source"})
elif refers(value["segments"]):
positions = [i for i, item in enumerate(value["segments"]) if refers(item)]
later = value["segments"][positions[0] + 1:]
# Sequence membership is not an input dependency for
# independently rendered pixel takes. Invalidate this
# sequence's manifest, never delete its surviving clips.
blocked = [item for item in later if not independent_successor(item)]
if blocked:
for item in blocked:
dependents.append({"scene": item.get("index"), "revision": item.get("revision"),
"metadata_path": item.get("revision_metadata") or self._address(path),
"reason": "saved branch continues after this take"})
else:
owned[path] = "Affected branch manifest (invalidated; assembled video kept)"
else:
other = value["segment"]
ignored = {"segment"}
if (refers(value.get("processing_lineage"))
and independent_successor(other)
and _independent_pixel_take(value)):
ignored.add("processing_lineage")
if (refers({k: v for k, v in value.items() if k not in ignored})
or refers(other)
or (not value.get("processing_lineage") and path.parent.parent == profile
and other["index"] > scene and segment.get("context_steps", 0))):
dependents.append({"scene": other["index"], "revision": other["revision"],
"metadata_path": other["revision_metadata"],
"reason": "saved take depends on this source or branch"})
dependents = list({item["metadata_path"]: item for item in dependents}.values())
for item in dependents:
retained.pop(item["metadata_path"], None)
files = []
for path, label in sorted(owned.items()):
self._path(self._address(path))
if path.exists() and not path.is_file():
raise ValueError("Expected a regular processing file: %s" % path.name)
stat = path.stat() if path.exists() else None
files.append({"path": self._address(path), "label": label, "owned": True,
"shared": False, "exists": stat is not None,
"size_bytes": stat.st_size if stat else 0,
"_identity": (stat.st_dev, stat.st_ino, stat.st_size, stat.st_mtime_ns,
stat.st_ctime_ns) if stat else None})
snapshot = hashlib.sha256(json.dumps({
"run": run, "target": address, "files": files,
"documents": {self._address(p): v for p, v in docs.items()},
}, sort_keys=True).encode()).hexdigest()
return {"ok": True, "run_name": run, "metadata_path": address,
"scene": scene, "revision": revision, "profile": profile.name,
"allowed": not dependents, "dependents": dependents,
"retained_independent_takes": sorted(retained.values(), key=lambda item: (
item["scene"], item["metadata_path"])),
"blockers": (["Delete the dependent processed takes first; their saved branch/source still uses this take."]
if dependents else []),
"files": [{k: v for k, v in f.items() if not k.startswith("_")} for f in files],
"owned_file_count": sum(f["exists"] for f in files),
"reclaimed_bytes": sum(f["size_bytes"] for f in files), "snapshot": snapshot,
"not_deleted": ["Original generation clips and checkpoints", "Shared references and reference caches",
"Other processed takes and profiles", "Assembled videos, run archives and prompt history"]}
def deletion_preview(self, run_name, metadata_path):
run = _strict_run_name(run_name)
with checkpoint_run_lock(str(self.root), run):
return self._preview(run, metadata_path)
def delete(self, run_name, metadata_path, expected_snapshot=""):
run = _strict_run_name(run_name)
with checkpoint_run_lock(str(self.root), run):
preview = self._preview(run, metadata_path)
if not preview["allowed"]:
raise CheckpointDeleteBlocked(" ".join(preview["blockers"]), preview)
if not expected_snapshot or expected_snapshot != preview["snapshot"]:
raise CheckpointDeleteBlocked("Processed files or dependencies changed; preview deletion again.", preview)
staged = []
transaction = uuid.uuid4().hex
try:
for item in preview["files"]:
if not item["exists"]:
continue
path = self._path(item["path"])
temporary = path.with_name(path.name + ".delete." + transaction + ".tmp")
os.replace(path, temporary)
staged.append((path, temporary, item["size_bytes"]))
except Exception:
for path, temporary, _ in reversed(staged):
os.replace(temporary, path)
raise
pending, reclaimed = [], 0
for path, temporary, size in staged:
try:
temporary.unlink()
reclaimed += size
except OSError:
pending.append(self._address(temporary))
return {"ok": True, "deleted_files": len(staged) - len(pending),
"reclaimed_bytes": reclaimed, "cleanup_pending": pending,
"message": "Deleted processed scene %d take %s (%s). Original clips are unchanged.%s" % (
preview["scene"], preview["revision"][:8], preview["profile"],
" Some staged files could not be removed; cleanup_pending lists their paths." if pending else "")}
def require_saved_processing_segments(output_root, segments):
"""Fence in-flight saves/loop completion against a concurrent deletion.
Called under checkpoint_run_lock, before publishing pointers or manifests.
JSON identity and file presence only; never load/hash multi-GB tensors here.
"""
manager = ProcessingCheckpointManager(output_root)
for segment in segments:
path = manager._path(segment.get("revision_metadata"))
if not path.is_file():
raise ValueError("A processed source/branch take was deleted while this run was executing; reselect the source and requeue.")
saved = manager._read(path).get("segment", {})
if any(saved.get(key) != segment.get(key) for key in (
"index", "revision", "checkpoint", "checkpoint_sha256")):
raise ValueError("Processed source/branch identity changed while this run was executing.")
if not manager._path(saved.get("checkpoint")).is_file():
raise ValueError("A processed source/branch checkpoint is missing; reselect the source and requeue.")
|