diff --git a/AGENT_GUIDE.md b/AGENT_GUIDE.md index e1506c06..488a03a1 100644 --- a/AGENT_GUIDE.md +++ b/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. diff --git a/lib/checkpoint.py b/lib/checkpoint.py index d8a37c88..b2b597a0 100644 --- a/lib/checkpoint.py +++ b/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// — 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 diff --git a/lib/events.py b/lib/events.py index a8449b53..a552b861 100644 --- a/lib/events.py +++ b/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: diff --git a/lib/paths.py b/lib/paths.py new file mode 100644 index 00000000..5e962685 --- /dev/null +++ b/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" diff --git a/lib/pipeline_loader.py b/lib/pipeline_loader.py index 727f6791..6ced59c7 100644 --- a/lib/pipeline_loader.py +++ b/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"]: diff --git a/skills/pipelines/documentary-montage/edit-director.md b/skills/pipelines/documentary-montage/edit-director.md index e26496ee..8468974f 100644 --- a/skills/pipelines/documentary-montage/edit-director.md +++ b/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. diff --git a/tools/base_tool.py b/tools/base_tool.py index 3440c1d7..1194e97d 100644 --- a/tools/base_tool.py +++ b/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]