fix: recover bounded defects from PR backlog

lib/checkpoint.py

@@ -95,9 +95,29 @@ class CheckpointValidationError(ValueError):
     """Raised when a checkpoint or its canonical artifacts are invalid."""
 
 
+def _validate_style_playbook(style_playbook: str | None) -> None:
+    """Fail closed when a checkpoint names a visual identity that cannot load."""
+
+    if style_playbook is None:
+        return
+    try:
+        from styles.playbook_loader import list_playbooks, load_playbook
+
+        load_playbook(style_playbook)
+    except Exception as exc:
+        try:
+            available = list_playbooks()
+        except Exception:
+            available = []
+        raise CheckpointValidationError(
+            f"Unknown or invalid style_playbook {style_playbook!r}. "
+            f"Available playbooks: {available}. Underlying error: {exc}"
+        ) from exc
+
+
 @lru_cache(maxsize=1)
 def _load_checkpoint_schema() -> dict[str, Any]:
-    with open(CHECKPOINT_SCHEMA_PATH) as f:
+    with open(CHECKPOINT_SCHEMA_PATH, encoding="utf-8") as f:
         return json.load(f)
 
 
@@ -192,6 +212,7 @@ def init_project(
     Idempotent: re-running preserves the original created_at and merges fields.
     Returns the project directory.
     """
+    _validate_style_playbook(style_playbook)
     base = pipeline_dir or PROJECTS_DIR
     project_dir = base / project_id
     for sub in (
@@ -208,7 +229,7 @@ def init_project(
     marker: dict[str, Any] = {}
     if marker_path.exists():
         try:
-            with open(marker_path) as f:
+            with open(marker_path, encoding="utf-8") as f:
                 marker = json.load(f)
         except (json.JSONDecodeError, OSError):
             marker = {}
@@ -221,7 +242,7 @@ def init_project(
     if style_playbook is not None:
         marker["style_playbook"] = style_playbook
 
-    with open(marker_path, "w") as f:
+    with open(marker_path, "w", encoding="utf-8") as f:
         json.dump(marker, f, indent=2)
 
     return project_dir
@@ -260,6 +281,71 @@ def _stage_requires_approval(pipeline_type: Optional[str], stage: str) -> Option
     return get_stage_human_approval_default(manifest, stage)
 
 
+def _enforce_stage_prerequisites(
+    pipeline_dir: Path,
+    project_id: str,
+    pipeline_type: str | None,
+    stage: str,
+    status: str,
+) -> None:
+    """Require completed, approved predecessors before advancing a stage.
+
+    ``in_progress`` and failure heartbeats remain writable so an operator can
+    inspect or resume a broken run. Only lifecycle advancement
+    (``awaiting_human``/``completed``) is gated.
+    """
+
+    if status not in {"awaiting_human", "completed"}:
+        return
+    if not pipeline_type or pipeline_type == "unknown":
+        return
+
+    stages = get_pipeline_stages(pipeline_type)
+    if stage not in stages:
+        return
+
+    incomplete: list[str] = []
+    unapproved: list[str] = []
+    for predecessor in stages[: stages.index(stage)]:
+        path = _checkpoint_path(pipeline_dir, project_id, predecessor)
+        if not path.exists():
+            incomplete.append(predecessor)
+            continue
+        try:
+            with open(path, encoding="utf-8") as handle:
+                checkpoint = json.load(handle)
+            validate_checkpoint(checkpoint)
+        except (OSError, json.JSONDecodeError, CheckpointValidationError):
+            incomplete.append(predecessor)
+            continue
+        if (
+            checkpoint.get("project_id") != project_id
+            or checkpoint.get("pipeline_type") != pipeline_type
+            or checkpoint.get("stage") != predecessor
+        ):
+            incomplete.append(predecessor)
+            continue
+        if checkpoint.get("status") != "completed":
+            incomplete.append(predecessor)
+            continue
+        if _stage_requires_approval(pipeline_type, predecessor) and not checkpoint.get(
+            "human_approved"
+        ):
+            unapproved.append(predecessor)
+
+    if incomplete or unapproved:
+        details = []
+        if incomplete:
+            details.append(f"incomplete or missing: {incomplete}")
+        if unapproved:
+            details.append(f"completed without required approval: {unapproved}")
+        raise CheckpointValidationError(
+            f"PREREQUISITE VIOLATION: stage {stage!r} cannot advance; "
+            + "; ".join(details)
+            + f". Pipeline order: {stages}."
+        )
+
+
 def _archive_superseded_checkpoint(path: Path, stage: str) -> None:
     """Copy an existing checkpoint into history/ before it is overwritten.
 
@@ -275,7 +361,7 @@ def _archive_superseded_checkpoint(path: Path, stage: str) -> None:
     if not path.exists():
         return
     try:
-        with open(path) as f:
+        with open(path, encoding="utf-8") as f:
             existing = json.load(f)
     except (json.JSONDecodeError, OSError):
         existing = {}
@@ -314,7 +400,7 @@ def _merge_decision_log(
     """
     path = _decision_log_path(pipeline_dir, project_id)
     if path.exists():
-        with open(path) as f:
+        with open(path, encoding="utf-8") as f:
             existing = json.load(f)
     else:
         existing = {
@@ -329,7 +415,7 @@ def _merge_decision_log(
             existing["decisions"].append(decision)
 
     path.parent.mkdir(parents=True, exist_ok=True)
-    with open(path, "w") as f:
+    with open(path, "w", encoding="utf-8") as f:
         json.dump(existing, f, indent=2)
 
 
@@ -351,19 +437,22 @@ def write_checkpoint(
     metadata: Optional[dict] = None,
 ) -> Path:
     """Write a checkpoint file for a pipeline stage."""
-    # Backfill a missing pipeline_type from the project marker so that
-    # omitting the kwarg doesn't quietly bypass gate enforcement.
-    if not pipeline_type:
-        marker = None
-        marker_path = pipeline_dir / project_id / PROJECT_MARKER_FILENAME
-        if marker_path.exists():
-            try:
-                with open(marker_path) as f:
-                    marker = json.load(f)
-            except (json.JSONDecodeError, OSError):
-                marker = None
-        if isinstance(marker, dict) and marker.get("pipeline_type"):
+    # Backfill identity fields from the project marker so omitted kwargs
+    # cannot bypass either gate enforcement or style validation.
+    marker = None
+    marker_path = pipeline_dir / project_id / PROJECT_MARKER_FILENAME
+    if marker_path.exists() and (not pipeline_type or not style_playbook):
+        try:
+            with open(marker_path, encoding="utf-8") as f:
+                marker = json.load(f)
+        except (json.JSONDecodeError, OSError):
+            marker = None
+    if isinstance(marker, dict):
+        if not pipeline_type and marker.get("pipeline_type"):
             pipeline_type = marker["pipeline_type"]
+        if not style_playbook and marker.get("style_playbook"):
+            style_playbook = marker["style_playbook"]
+    _validate_style_playbook(style_playbook)
 
     valid_stages = (
         set(get_pipeline_stages(pipeline_type)) if pipeline_type
@@ -404,6 +493,14 @@ def write_checkpoint(
                 f"re-write with status='completed', human_approved=True."
             )
 
+    _enforce_stage_prerequisites(
+        pipeline_dir,
+        project_id,
+        pipeline_type,
+        stage,
+        status,
+    )
+
     checkpoint = {
         "version": "1.0",
         "project_id": project_id,
@@ -456,7 +553,7 @@ def write_checkpoint(
     # current checkpoint; then archive the superseded file and swap in the
     # new one atomically.
     tmp_path = path.with_suffix(".json.tmp")
-    with open(tmp_path, "w") as f:
+    with open(tmp_path, "w", encoding="utf-8") as f:
         json.dump(checkpoint, f, indent=2)
     # Preserve run history: a superseded completed/awaiting_human checkpoint
     # is copied to history/ (stage versioning, gate audit trail, replay).
@@ -474,7 +571,7 @@ def read_checkpoint(
     path = _checkpoint_path(pipeline_dir, project_id, stage)
     if not path.exists():
         return None
-    with open(path) as f:
+    with open(path, encoding="utf-8") as f:
         checkpoint = json.load(f)
     validate_checkpoint(checkpoint)
     return checkpoint
@@ -496,7 +593,7 @@ def get_latest_checkpoint(
     if not checkpoints:
         return None
 
-    with open(checkpoints[0]) as f:
+    with open(checkpoints[0], encoding="utf-8") as f:
         checkpoint = json.load(f)
     validate_checkpoint(checkpoint)
     return checkpoint

lib/clip_embedder.py

@@ -61,6 +61,20 @@ def model_info() -> dict:
     }
 
 
+def _as_feature_tensor(features):
+    """Normalize CLIP feature return values across transformers versions.
+
+    Transformers 4 returned the projected tensor directly. Transformers 5 may
+    wrap that tensor in a model-output object whose ``pooler_output`` contains
+    the same shared-space embedding. Do not project it again: the vision
+    projection expects the pre-projection width, while ``pooler_output`` is
+    already the final CLIP width.
+    """
+
+    pooled = getattr(features, "pooler_output", None)
+    return features if pooled is None else pooled
+
+
 def embed_images(image_paths: Sequence[Union[str, Path]]) -> np.ndarray:
     """Embed a list of image files into a (N, 512) float32 matrix.
 
@@ -82,7 +96,7 @@ def embed_images(image_paths: Sequence[Union[str, Path]]) -> np.ndarray:
 
     inputs = _PROCESSOR(images=images, return_tensors="pt").to(_DEVICE)
     with torch.no_grad():
-        features = _MODEL.get_image_features(**inputs)
+        features = _as_feature_tensor(_MODEL.get_image_features(**inputs))
     features = features / features.norm(dim=-1, keepdim=True).clamp_min(1e-8)
     arr = features.cpu().numpy().astype(np.float32, copy=False)
     # Close PIL handles to avoid leaking file handles on Windows
@@ -116,7 +130,7 @@ def embed_texts(texts: Sequence[str]) -> np.ndarray:
         max_length=77,
     ).to(_DEVICE)
     with torch.no_grad():
-        features = _MODEL.get_text_features(**inputs)
+        features = _as_feature_tensor(_MODEL.get_text_features(**inputs))
     features = features / features.norm(dim=-1, keepdim=True).clamp_min(1e-8)
     return features.cpu().numpy().astype(np.float32, copy=False)
 

lib/pipeline_loader.py

@@ -26,7 +26,7 @@ from functools import lru_cache
 
 @lru_cache(maxsize=1)
 def _load_manifest_schema() -> dict:
-    with open(SCHEMA_PATH) as f:
+    with open(SCHEMA_PATH, encoding="utf-8") as f:
         return json.load(f)
 
 
@@ -61,7 +61,7 @@ def load_pipeline(name: str, defs_dir: Optional[Path] = None) -> dict[str, Any]:
     if not path.exists():
         raise FileNotFoundError(f"Pipeline manifest not found: {path}")
 
-    with open(path) as f:
+    with open(path, encoding="utf-8") as f:
         manifest = yaml.safe_load(f)
 
     schema = _load_manifest_schema()