backlot phase 0 review fixes: fail-closed gates, atomic checkpoint writes, event attribution hardening

- gate = manifest OR caller (stricter wins); unknown pipeline_type now
  raises instead of silently disabling enforcement; corrupt manifest logs
  and falls back; misleading diagnostic fixed
- write_checkpoint backfills pipeline_type from project.json marker so
  omitting the kwarg can't bypass gates
- checkpoint writes are atomic (temp + os.replace); history archiving is
  copy-based and best-effort (Windows open-file safe)
- manifest loads cached (load_pipeline_readonly); stage gate lookup moved
  to pipeline_loader.get_stage_human_approval_default; PROJECTS_DIR unified
  in lib/paths.py
- events: containment + root-normalization for explicit project dirs, no
  ghost-project mkdir, 0.0 cost preserved, nested-call depth tag,
  wrapper simplified
- documentary-montage edit-director gate footer (was missed); AGENT_GUIDE
  no longer claims edit/compose always auto-proceed

AGENT_GUIDE.md

@@ -576,7 +576,7 @@ The reviewer is a meta skill (`skills/meta/reviewer.md`) — advisory, never dir
 The checkpoint protocol meta skill (`skills/meta/checkpoint-protocol.md`) teaches the agent when to pause:
 
 - Read `human_approval_default` from the pipeline manifest per stage. **The manifest value is binding** — never re-judge it. `lib/checkpoint.py` enforces this: a gated stage cannot be written `completed` without `human_approved=True`.
-- Gated stages across all pipelines: `idea`/`proposal`, `script`, `scene_plan`, **`assets`** (review the generated assets scene-by-scene — the Backlot board's filmstrip — before compose locks them in), and `publish`. `edit` and `compose` auto-proceed.
+- Typical gated stages: `idea`/`proposal`, `script`, `scene_plan`, **`assets`** (review the generated assets scene-by-scene — the Backlot board's filmstrip — before compose locks them in), and `publish` where the pipeline has one. Most pipelines auto-proceed on `edit` and `compose`, but not all (documentary-montage gates `edit`) — the manifest you loaded is the only authority.
 - When approval is required: write the checkpoint as `awaiting_human`, present artifact summary, review findings, and cost snapshot — then **END YOUR TURN**. Doing further pipeline work in the same response is a gate violation.
 - **Approval is per-gate.** An early "go ahead" never covers later gates; explicit full-run pre-authorization must be recorded as a `decision_log` entry (`category: "approval_policy"`) to count.
 - Wait for human to approve, request revision, or abort.

lib/checkpoint.py

@@ -67,8 +67,8 @@ def get_pipeline_stages(pipeline_type: str | None) -> list[str]:
         return list(STAGES)
 
     try:
-        from lib.pipeline_loader import load_pipeline, get_stage_order
-        manifest = load_pipeline(pipeline_type)
+        from lib.pipeline_loader import load_pipeline_readonly, get_stage_order
+        manifest = load_pipeline_readonly(pipeline_type)
         return get_stage_order(manifest)
     except (FileNotFoundError, Exception):
         # Graceful fallback: return all known stages in canonical order
@@ -85,7 +85,7 @@ CHECKPOINT_SCHEMA_PATH = (
 # live under PROJECTS_DIR/<project_id>/ — this is the location the Backlot
 # board watches. Callers may still pass a different pipeline_dir (tests do),
 # but production runs should use the default.
-PROJECTS_DIR = Path(__file__).resolve().parent.parent / "projects"
+from lib.paths import PROJECTS_DIR  # noqa: E402  (single source of truth)
 
 PROJECT_MARKER_FILENAME = "project.json"
 HISTORY_DIRNAME = "history"
@@ -221,30 +221,47 @@ def init_project(
 def _stage_requires_approval(pipeline_type: Optional[str], stage: str) -> Optional[bool]:
     """Read human_approval_default for a stage from its pipeline manifest.
 
-    Returns None when the manifest can't answer (unknown pipeline_type,
-    stage not declared) — the caller then falls back to the value the
+    Returns None when the stage isn't declared in the manifest or no
+    pipeline_type was given — the caller then falls back to the value the
     agent passed in.
+
+    A *provided but unknown* pipeline_type raises: a typo must not silently
+    disable gate enforcement (fail-closed, not fail-open). Other manifest
+    load failures are logged and fall back — a corrupt manifest shouldn't
+    strand an otherwise-valid run, but the degradation must be visible.
     """
     if not pipeline_type or pipeline_type == "unknown":
         return None
+    from lib.pipeline_loader import get_stage_human_approval_default, load_pipeline_readonly
     try:
-        from lib.pipeline_loader import load_pipeline
-        manifest = load_pipeline(pipeline_type)
-        for stage_def in manifest.get("stages", []):
-            if stage_def.get("name") == stage:
-                return bool(stage_def.get("human_approval_default", False))
-    except Exception:
+        manifest = load_pipeline_readonly(pipeline_type)
+    except FileNotFoundError:
+        raise CheckpointValidationError(
+            f"Unknown pipeline_type {pipeline_type!r} — cannot resolve gate "
+            f"policy for stage {stage!r}. Check the spelling against "
+            f"pipeline_defs/*.yaml."
+        )
+    except Exception as exc:
+        import logging
+        logging.getLogger(__name__).warning(
+            "Gate policy unavailable for pipeline %r (%s) — falling back to "
+            "the caller's human_approval_required flag.", pipeline_type, exc,
+        )
         return None
-    return None
+    return get_stage_human_approval_default(manifest, stage)
 
 
 def _archive_superseded_checkpoint(path: Path, stage: str) -> None:
-    """Move an existing checkpoint into history/ before it is overwritten.
+    """Copy an existing checkpoint into history/ before it is overwritten.
 
     Preserves the full run record: stage re-runs (script v1 → v2) and gate
     transitions (awaiting_human → completed) remain reconstructable. Repeated
     in_progress refreshes are NOT archived — they are partial-progress
     heartbeats, not versions.
+
+    Archiving is best-effort and must never crash a checkpoint write: the
+    Backlot watcher may hold the file open (Windows denies renames of open
+    files), so we copy rather than move, and swallow archival I/O failures.
     """
     if not path.exists():
         return
@@ -256,16 +273,21 @@ def _archive_superseded_checkpoint(path: Path, stage: str) -> None:
     if existing.get("status") == "in_progress":
         return
 
-    stamp = str(existing.get("timestamp", ""))
-    safe_stamp = "".join(c for c in stamp if c.isalnum()) or f"{path.stat().st_mtime_ns}"
-    history_dir = path.parent / HISTORY_DIRNAME
-    history_dir.mkdir(parents=True, exist_ok=True)
-    target = history_dir / f"checkpoint_{stage}_{safe_stamp}.json"
-    counter = 1
-    while target.exists():
-        target = history_dir / f"checkpoint_{stage}_{safe_stamp}_{counter}.json"
-        counter += 1
-    path.replace(target)
+    try:
+        import shutil
+        stamp = str(existing.get("timestamp", ""))
+        safe_stamp = "".join(c for c in stamp if c.isalnum()) or f"{path.stat().st_mtime_ns}"
+        history_dir = path.parent / HISTORY_DIRNAME
+        history_dir.mkdir(parents=True, exist_ok=True)
+        target = history_dir / f"checkpoint_{stage}_{safe_stamp}.json"
+        if target.exists():
+            target = history_dir / f"checkpoint_{stage}_{safe_stamp}_{path.stat().st_mtime_ns}.json"
+        shutil.copyfile(path, target)
+    except OSError:
+        import logging
+        logging.getLogger(__name__).warning(
+            "Could not archive superseded checkpoint %s to history/", path
+        )
 
 
 def _decision_log_path(pipeline_dir: Path, project_id: str) -> Path:
@@ -320,6 +342,20 @@ 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"):
+            pipeline_type = marker["pipeline_type"]
+
     valid_stages = (
         set(get_pipeline_stages(pipeline_type)) if pipeline_type
         else ALL_KNOWN_STAGES
@@ -332,22 +368,31 @@ def write_checkpoint(
 
     # --- Gate enforcement (GI-4) ---
     # The pipeline manifest is the binding source of truth for whether a stage
-    # gates on human approval. A gated stage can only be written as
-    # "completed" with explicit evidence of approval (human_approved=True).
-    # Skipping a gate is a hard error, not a soft violation.
+    # gates on human approval; a caller may gate MORE strictly (e.g. a
+    # manual_all checkpoint policy) but never less. A gated stage can only be
+    # written "completed" with explicit evidence of approval
+    # (human_approved=True). Skipping a gate is a hard error.
+    #
+    # Enforcement happens at write time only: pre-existing checkpoints written
+    # before gating (or by hand) still read as completed — deliberate
+    # back-compat so in-flight and legacy projects keep resuming.
     manifest_gate = _stage_requires_approval(pipeline_type, stage)
-    gated = manifest_gate if manifest_gate is not None else human_approval_required
+    gated = bool(manifest_gate) or human_approval_required
     if gated:
         human_approval_required = True
         if status == "completed" and not human_approved:
+            gate_source = (
+                f"human_approval_default: true in the {pipeline_type!r} manifest"
+                if manifest_gate
+                else "human_approval_required=True was passed by the caller"
+            )
             raise CheckpointValidationError(
                 f"GATE VIOLATION: stage {stage!r} requires human approval "
-                f"(human_approval_default: true in the {pipeline_type!r} manifest) "
-                f"but status='completed' was written without human_approved=True. "
-                f"Correct protocol: write status='awaiting_human', present the "
-                f"artifact summary to the user, END YOUR TURN, and only after "
-                f"the user approves re-write with status='completed', "
-                f"human_approved=True."
+                f"({gate_source}) but status='completed' was written without "
+                f"human_approved=True. Correct protocol: write "
+                f"status='awaiting_human', present the artifact summary to the "
+                f"user, END YOUR TURN, and only after the user approves "
+                f"re-write with status='completed', human_approved=True."
             )
 
     checkpoint = {
@@ -397,12 +442,18 @@ def write_checkpoint(
 
     path = _checkpoint_path(pipeline_dir, project_id, stage)
     path.parent.mkdir(parents=True, exist_ok=True)
-    # Preserve run history: a superseded completed/awaiting_human checkpoint
-    # moves to history/ instead of being destroyed (stage versioning, gate
-    # audit trail, replay).
-    _archive_superseded_checkpoint(path, stage)
-    with open(path, "w") as f:
+    # Serialize to a temp file first so a mid-write failure (disk full,
+    # unserializable metadata) can never leave the stage with a truncated
+    # 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:
         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).
+    _archive_superseded_checkpoint(path, stage)
+    import os
+    os.replace(tmp_path, path)
 
     return path
 

lib/events.py

@@ -19,11 +19,13 @@ from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any, Optional
 
-REPO_ROOT = Path(__file__).resolve().parent.parent
-PROJECTS_DIR = REPO_ROOT / "projects"
+from lib.paths import PROJECTS_DIR, REPO_ROOT  # single source of truth
 
 EVENTS_FILENAME = "events.jsonl"
 
+# Thread-level serialization only. Cross-PROCESS appends are unsynchronized
+# by design: single-line O_APPEND writes rarely tear, and read_events skips
+# malformed lines, so a torn line degrades to one missing activity entry.
 _write_lock = threading.Lock()
 
 # Input keys checked (in order) when inferring the project a tool call
@@ -50,14 +52,13 @@ def infer_project_dir(inputs: Any) -> Optional[Path]:
     if not isinstance(inputs, dict):
         return None
     try:
-        for key in _EXPLICIT_PROJECT_KEYS:
-            value = inputs.get(key)
-            if isinstance(value, (str, Path)) and str(value):
-                p = Path(value)
-                if p.is_dir():
-                    return p
+        # Only paths under the canonical projects root are attributable —
+        # an explicit project_dir pointing elsewhere (HyperFrames workspace,
+        # arbitrary user dir) must not receive an events.jsonl. Explicit
+        # values are normalized to the project ROOT the same way hints are,
+        # so project_dir="projects/x/renders/build" attributes to projects/x.
         projects_root = PROJECTS_DIR.resolve()
-        for key in _PATH_HINT_KEYS:
+        for key in _EXPLICIT_PROJECT_KEYS + _PATH_HINT_KEYS:
             value = inputs.get(key)
             if not isinstance(value, (str, Path)) or not str(value):
                 continue
@@ -74,12 +75,18 @@ def infer_project_dir(inputs: Any) -> Optional[Path]:
 
 
 def emit_event(project_dir: Path | str, payload: dict[str, Any]) -> None:
-    """Append one event to the project's events.jsonl. Never raises."""
+    """Append one event to the project's events.jsonl. Never raises.
+
+    Writes only into an EXISTING project directory — a typo'd path must not
+    spawn a ghost project on the board.
+    """
     try:
+        project_dir = Path(project_dir)
+        if not project_dir.is_dir():
+            return
         entry = {"ts": datetime.now(timezone.utc).isoformat()}
         entry.update({k: v for k, v in payload.items() if v is not None})
-        path = Path(project_dir) / EVENTS_FILENAME
-        path.parent.mkdir(parents=True, exist_ok=True)
+        path = project_dir / EVENTS_FILENAME
         line = json.dumps(entry, default=str)
         with _write_lock:
             with open(path, "a", encoding="utf-8") as f:

lib/paths.py

@@ -0,0 +1,13 @@
+"""Canonical repository paths — single source of truth.
+
+The projects root is the most load-bearing path in the system: checkpoints
+are written under it, tool events are attributed against it, and the Backlot
+board watches it. Define it once.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+PROJECTS_DIR = REPO_ROOT / "projects"

lib/pipeline_loader.py

@@ -21,11 +21,31 @@ SCHEMA_PATH = (
 )
 
 
+from functools import lru_cache
+
+
+@lru_cache(maxsize=1)
 def _load_manifest_schema() -> dict:
     with open(SCHEMA_PATH) as f:
         return json.load(f)
 
 
+@lru_cache(maxsize=64)
+def _load_pipeline_cached(name: str, defs_dir_key: str) -> dict[str, Any]:
+    """Cached manifest load. Treat the returned dict as READ-ONLY."""
+    return load_pipeline(name, Path(defs_dir_key) if defs_dir_key else None)
+
+
+def load_pipeline_readonly(name: str, defs_dir: Optional[Path] = None) -> dict[str, Any]:
+    """Load a manifest through a cache. The result MUST NOT be mutated.
+
+    Manifests are immutable within a run; hot paths (gate checks on every
+    checkpoint write, board state derivation) should use this instead of
+    re-parsing YAML + re-validating the schema each call.
+    """
+    return _load_pipeline_cached(name, str(defs_dir) if defs_dir else "")
+
+
 def load_pipeline(name: str, defs_dir: Optional[Path] = None) -> dict[str, Any]:
     """Load and validate a pipeline manifest by name.
 
@@ -150,6 +170,18 @@ def get_stage_skill(manifest: dict, stage_name: str) -> Optional[str]:
     return None
 
 
+def get_stage_human_approval_default(manifest: dict, stage_name: str) -> Optional[bool]:
+    """Whether a stage gates on human approval. None if the stage isn't declared.
+
+    This is the single lookup used by gate enforcement (lib/checkpoint.py)
+    and the Backlot board — keep them reading the same field the same way.
+    """
+    for stage in manifest["stages"]:
+        if stage["name"] == stage_name:
+            return bool(stage.get("human_approval_default", False))
+    return None
+
+
 def get_stage_review_focus(manifest: dict, stage_name: str) -> list[str]:
     """Get the review focus items for a stage."""
     for stage in manifest["stages"]:

skills/pipelines/documentary-montage/edit-director.md

@@ -369,3 +369,12 @@ Canonical shape for this pipeline:
 This gives a 90s piece with 3 breathing points (fade_in, silence,
 fade_out), a clear hero arc (slots 1 → 11 → 15), and no adjacent
 scale collisions.
+
+---
+
+## Gate Reminder (Binding)
+
+This stage gates on human approval (`human_approval_default: true`). After review passes:
+checkpoint with `status="awaiting_human"`, present the summary (the Backlot board renders
+the artifact), and **END YOUR TURN**. Do not start the next stage in the same response.
+Approval is per-gate — an earlier "go ahead" does not cover this gate.

tools/base_tool.py

@@ -138,6 +138,13 @@ class ToolResult:
     model: Optional[str] = None
 
 
+import threading as _threading
+
+# Shared nesting counter for instrumented execute() calls (thread-local so
+# parallel tool threads don't see each other's depth).
+_EXECUTE_DEPTH = _threading.local()
+
+
 def _instrument_execute(fn: Callable) -> Callable:
     """Wrap a tool's execute() with Backlot event emission.
 
@@ -152,57 +159,65 @@ def _instrument_execute(fn: Callable) -> Callable:
     if getattr(fn, "_backlot_instrumented", False):
         return fn
 
+    depth_state = _EXECUTE_DEPTH  # shared across all tools (selector → provider)
+
     @functools.wraps(fn)
     def wrapper(self, inputs: Any, *args: Any, **kwargs: Any):
-        project_dir = None
+        # Event layer is fully optional: if it can't import, run untouched.
+        try:
+            from lib.events import emit_event, infer_project_dir
+        except Exception:
+            return fn(self, inputs, *args, **kwargs)
+
         tool_name = getattr(self, "name", "") or self.__class__.__name__
         scene_id = inputs.get("scene_id") if isinstance(inputs, dict) else None
         output_path = inputs.get("output_path") if isinstance(inputs, dict) else None
-        try:
-            from lib.events import emit_event, infer_project_dir
-            project_dir = infer_project_dir(inputs)
-            if project_dir is not None:
-                emit_event(project_dir, {
-                    "tool": tool_name,
-                    "event": "start",
-                    "scene_id": scene_id,
-                    "output_path": str(output_path) if output_path else None,
-                })
-        except Exception:
-            project_dir = None
+        # Nesting depth: selector tools delegate to provider tools' execute().
+        # Both emit (the ticker wants the provider name too), but depth lets
+        # consumers dedupe — e.g. sum cost_usd only at depth 0.
+        depth = getattr(depth_state, "value", 0)
+        depth_state.value = depth + 1
+        project_dir = infer_project_dir(inputs)
+
+        base = {
+            "tool": tool_name,
+            "scene_id": scene_id,
+            "depth": depth if depth else None,
+        }
+        if project_dir is not None:
+            emit_event(project_dir, {
+                **base, "event": "start",
+                "output_path": str(output_path) if output_path else None,
+            })
 
         started = time.monotonic()
         try:
             result = fn(self, inputs, *args, **kwargs)
         except Exception as exc:
             if project_dir is not None:
-                try:
-                    from lib.events import emit_event
-                    emit_event(project_dir, {
-                        "tool": tool_name,
-                        "event": "error",
-                        "scene_id": scene_id,
-                        "error": str(exc)[:300],
-                        "duration_s": round(time.monotonic() - started, 2),
-                    })
-                except Exception:
-                    pass
-            raise
-
-        if project_dir is not None:
-            try:
-                from lib.events import emit_event
                 emit_event(project_dir, {
-                    "tool": tool_name,
-                    "event": "finish",
-                    "scene_id": scene_id,
-                    "output_path": str(output_path) if output_path else None,
-                    "success": getattr(result, "success", None),
-                    "cost_usd": getattr(result, "cost_usd", None) or None,
+                    **base, "event": "error",
+                    "error": str(exc)[:300],
                     "duration_s": round(time.monotonic() - started, 2),
                 })
-            except Exception:
-                pass
+            raise
+        finally:
+            depth_state.value = depth
+
+        if project_dir is None:
+            # The tool may have created its own project dir during execute
+            # (first call of a run) — attribute the finish if possible.
+            project_dir = infer_project_dir(inputs)
+        if project_dir is not None:
+            cost = getattr(result, "cost_usd", None)
+            emit_event(project_dir, {
+                **base, "event": "finish",
+                "output_path": str(output_path) if output_path else None,
+                "success": getattr(result, "success", None),
+                # NOTE: 0.0 is meaningful (ran for free) — only None is dropped.
+                "cost_usd": cost if isinstance(cost, (int, float)) else None,
+                "duration_s": round(time.monotonic() - started, 2),
+            })
         return result
 
     wrapper._backlot_instrumented = True  # type: ignore[attr-defined]