| """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 |
| 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:] |
| |
| |
| |
| 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.") |
|
|
