Repository navigation
Expand file tree
/
Copy pathcheckpoint.py
More file actions
536 lines (457 loc) · 19.9 KB
/
Copy pathcheckpoint.py
File metadata and controls
536 lines (457 loc) · 19.9 KB
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
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
"""Checkpoint writer/reader for pipeline state persistence.
Each stage writes a checkpoint after completion. The orchestrator uses
checkpoints to resume pipelines and to present state at human checkpoints.
"""
from __future__ import annotations
import json
from functools import lru_cache
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Optional
import jsonschema
from schemas.artifacts import ARTIFACT_NAMES, validate_artifact
# All known stages across all pipelines (used only for artifact name lookup).
ALL_KNOWN_STAGES = frozenset([
"research", "proposal", "idea", "script", "scene_plan",
"assets", "edit", "compose", "publish",
])
# Backward-compatible alias — existing code / tests that import STAGES still work.
# New code should use get_pipeline_stages(pipeline_type) instead.
STAGES = ["research", "proposal", "idea", "script", "scene_plan",
"assets", "edit", "compose", "publish"]
CANONICAL_STAGE_ARTIFACTS = {
"research": "research_brief",
"proposal": "proposal_packet",
"idea": "brief",
"script": "script",
"scene_plan": "scene_plan",
"assets": "asset_manifest",
"edit": "edit_decisions",
"compose": "render_report",
"publish": "publish_log",
}
# Additional artifacts that may be produced alongside canonical ones.
# These are not stage-defining but are required by governance contracts.
SUPPLEMENTARY_ARTIFACTS = {
"source_media_review", # Required before first planning stage when user media exists
"final_review", # Required by compose stage before presenting to user
"video_analysis_brief", # Reference-video grounding artifact carried alongside stages
}
def get_pipeline_stages(pipeline_type: str | None) -> list[str]:
"""Return the ordered stage list for a specific pipeline.
Falls back to STAGES (deterministic canonical order) when pipeline_type
is not provided or the manifest cannot be loaded.
Previous versions used a set intersection here, which produced
nondeterministic ordering. The fallback now uses a stable list.
"""
if pipeline_type is None:
# Deterministic canonical fallback — sorted to ensure stable ordering
import logging
logging.getLogger(__name__).warning(
"get_pipeline_stages called without pipeline_type — "
"using canonical fallback order. Pass pipeline_type for correctness."
)
return list(STAGES)
try:
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
return list(STAGES)
CHECKPOINT_SCHEMA_PATH = (
Path(__file__).resolve().parent.parent
/ "schemas"
/ "checkpoints"
/ "checkpoint.schema.json"
)
# Canonical project root. Checkpoints, artifacts, and the project marker all
# 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.
from lib.paths import PROJECTS_DIR # noqa: E402 (single source of truth)
PROJECT_MARKER_FILENAME = "project.json"
HISTORY_DIRNAME = "history"
class CheckpointValidationError(ValueError):
"""Raised when a checkpoint or its canonical artifacts are invalid."""
@lru_cache(maxsize=1)
def _load_checkpoint_schema() -> dict[str, Any]:
with open(CHECKPOINT_SCHEMA_PATH) as f:
return json.load(f)
def _validate_artifacts_for_stage(
stage: str,
status: str,
artifacts: dict[str, Any],
) -> None:
# Valid stages come from the pipeline manifest (get_pipeline_stages), which
# can declare stages beyond the 9 canonical ones (e.g. character-animation's
# `character_design`/`rig_plan`, screen-demo's `real_capture`). Those have no
# canonical artifact, so look it up defensively — a missing entry means the
# stage simply has no required artifact, not a crash.
required_artifact = CANONICAL_STAGE_ARTIFACTS.get(stage)
if (
required_artifact is not None
and status in {"completed", "awaiting_human"}
and required_artifact not in artifacts
):
raise CheckpointValidationError(
f"Stage {stage!r} with status {status!r} must include "
f"canonical artifact {required_artifact!r}"
)
for artifact_name, artifact_data in artifacts.items():
if artifact_name not in ARTIFACT_NAMES:
continue
if not isinstance(artifact_data, dict):
raise CheckpointValidationError(
f"Artifact {artifact_name!r} must be a JSON object matching its schema"
)
try:
validate_artifact(artifact_name, artifact_data)
except Exception as exc:
raise CheckpointValidationError(
f"Artifact {artifact_name!r} failed schema validation: {exc}"
) from exc
def validate_checkpoint(checkpoint: dict[str, Any]) -> None:
"""Validate checkpoint structure and canonical artifact payloads.
Uses pipeline_type (if present) to resolve the valid stage list.
Falls back to ALL_KNOWN_STAGES when pipeline_type is absent.
"""
stage = checkpoint.get("stage")
status = checkpoint.get("status")
artifacts = checkpoint.get("artifacts")
pipeline_type = checkpoint.get("pipeline_type")
valid_stages = (
set(get_pipeline_stages(pipeline_type)) if pipeline_type
else ALL_KNOWN_STAGES
)
if not isinstance(stage, str) or stage not in valid_stages:
raise CheckpointValidationError(
f"Invalid stage: {stage!r} for pipeline {pipeline_type!r}. "
f"Valid stages: {sorted(valid_stages)}"
)
if not isinstance(status, str):
raise CheckpointValidationError(f"Invalid status: {status!r}")
if not isinstance(artifacts, dict):
raise CheckpointValidationError("Checkpoint artifacts must be a dictionary")
_validate_artifacts_for_stage(stage, status, artifacts)
try:
jsonschema.validate(instance=checkpoint, schema=_load_checkpoint_schema())
except jsonschema.ValidationError as exc:
raise CheckpointValidationError(f"Checkpoint failed schema validation: {exc.message}") from exc
def _checkpoint_path(pipeline_dir: Path, project_id: str, stage: str) -> Path:
return pipeline_dir / project_id / f"checkpoint_{stage}.json"
def init_project(
project_id: str,
*,
title: str,
pipeline_type: str,
pipeline_dir: Optional[Path] = None,
style_playbook: Optional[str] = None,
) -> Path:
"""Initialize a project workspace with the canonical layout + marker file.
Creates projects/<project_id>/ with the standard subdirectories and writes
project.json — the marker the Backlot board uses to render a project's
identity and stage rail before the first checkpoint exists.
Idempotent: re-running preserves the original created_at and merges fields.
Returns the project directory.
"""
base = pipeline_dir or PROJECTS_DIR
project_dir = base / project_id
for sub in (
"artifacts",
"assets/images",
"assets/video",
"assets/audio",
"assets/music",
"renders",
):
(project_dir / sub).mkdir(parents=True, exist_ok=True)
marker_path = project_dir / PROJECT_MARKER_FILENAME
marker: dict[str, Any] = {}
if marker_path.exists():
try:
with open(marker_path) as f:
marker = json.load(f)
except (json.JSONDecodeError, OSError):
marker = {}
marker.setdefault("version", "1.0")
marker.setdefault("created_at", datetime.now(timezone.utc).isoformat())
marker["project_id"] = project_id
marker["title"] = title
marker["pipeline_type"] = pipeline_type
if style_playbook is not None:
marker["style_playbook"] = style_playbook
with open(marker_path, "w") as f:
json.dump(marker, f, indent=2)
return project_dir
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 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:
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 get_stage_human_approval_default(manifest, stage)
def _archive_superseded_checkpoint(path: Path, stage: str) -> None:
"""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
try:
with open(path) as f:
existing = json.load(f)
except (json.JSONDecodeError, OSError):
existing = {}
if existing.get("status") == "in_progress":
return
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:
return pipeline_dir / project_id / "decision_log.json"
def _merge_decision_log(
pipeline_dir: Path, project_id: str, new_log: dict[str, Any]
) -> None:
"""Append new decisions to the project-level decision log.
Each stage may produce decisions. This function merges them into a
single cumulative file so reviewers and the bench can inspect the
full audit trail.
"""
path = _decision_log_path(pipeline_dir, project_id)
if path.exists():
with open(path) as f:
existing = json.load(f)
else:
existing = {
"version": "1.0",
"project_id": project_id,
"decisions": [],
}
existing_ids = {d["decision_id"] for d in existing.get("decisions", [])}
for decision in new_log.get("decisions", []):
if decision.get("decision_id") not in existing_ids:
existing["decisions"].append(decision)
path.parent.mkdir(parents=True, exist_ok=True)
with open(path, "w") as f:
json.dump(existing, f, indent=2)
def write_checkpoint(
pipeline_dir: Path,
project_id: str,
stage: str,
status: str,
artifacts: dict[str, Any],
*,
pipeline_type: Optional[str] = None,
style_playbook: Optional[str] = None,
checkpoint_policy: str = "guided",
human_approval_required: bool = False,
human_approved: bool = False,
review: Optional[dict] = None,
cost_snapshot: Optional[dict] = None,
error: Optional[str] = None,
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
)
if stage not in valid_stages:
raise ValueError(
f"Invalid stage: {stage!r} for pipeline {pipeline_type!r}. "
f"Valid stages: {sorted(valid_stages)}"
)
# --- Gate enforcement (GI-4) ---
# The pipeline manifest is the binding source of truth for whether a stage
# 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 = 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"({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 = {
"version": "1.0",
"project_id": project_id,
"pipeline_type": pipeline_type or "unknown",
"stage": stage,
"status": status,
"timestamp": datetime.now(timezone.utc).isoformat(),
"checkpoint_policy": checkpoint_policy,
"human_approval_required": human_approval_required,
"human_approved": human_approved,
"artifacts": artifacts,
}
if style_playbook is not None:
checkpoint["style_playbook"] = style_playbook
if review is not None:
checkpoint["review"] = review
if cost_snapshot is not None:
checkpoint["cost_snapshot"] = cost_snapshot
if error is not None:
checkpoint["error"] = error
if metadata is not None:
checkpoint["metadata"] = metadata
# Merge decision_log: if this checkpoint carries new decisions,
# append them to the project-level decision log file, then write the
# reference back into relevant artifacts so downstream consumers can find it.
if "decision_log" in artifacts and isinstance(artifacts["decision_log"], dict):
_merge_decision_log(pipeline_dir, project_id, artifacts["decision_log"])
log_ref = str(_decision_log_path(pipeline_dir, project_id))
# Write decision_log_ref into proposal_packet and render_report
# artifacts if they are present in this checkpoint.
for artifact_key in ("proposal_packet", "render_report"):
if artifact_key in artifacts and isinstance(artifacts[artifact_key], dict):
plan_or_top = artifacts[artifact_key]
# proposal_packet stores it under production_plan
if artifact_key == "proposal_packet":
plan = plan_or_top.get("production_plan")
if isinstance(plan, dict):
plan["decision_log_ref"] = log_ref
else:
plan_or_top["decision_log_ref"] = log_ref
validate_checkpoint(checkpoint)
path = _checkpoint_path(pipeline_dir, project_id, stage)
path.parent.mkdir(parents=True, exist_ok=True)
# 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
def read_checkpoint(
pipeline_dir: Path, project_id: str, stage: str
) -> Optional[dict[str, Any]]:
"""Read a checkpoint file. Returns None if not found."""
path = _checkpoint_path(pipeline_dir, project_id, stage)
if not path.exists():
return None
with open(path) as f:
checkpoint = json.load(f)
validate_checkpoint(checkpoint)
return checkpoint
def get_latest_checkpoint(
pipeline_dir: Path, project_id: str
) -> Optional[dict[str, Any]]:
"""Find the most recent checkpoint for a project (by file mtime)."""
project_dir = pipeline_dir / project_id
if not project_dir.exists():
return None
checkpoints = sorted(
project_dir.glob("checkpoint_*.json"),
key=lambda p: p.stat().st_mtime,
reverse=True,
)
if not checkpoints:
return None
with open(checkpoints[0]) as f:
checkpoint = json.load(f)
validate_checkpoint(checkpoint)
return checkpoint
def get_completed_stages(
pipeline_dir: Path, project_id: str, pipeline_type: str | None = None
) -> list[str]:
"""Return list of stages that have a completed checkpoint.
When pipeline_type is provided, only checks stages defined in that
pipeline's manifest — preventing false positives from leftover
checkpoints of a different pipeline type.
"""
stages_to_check = get_pipeline_stages(pipeline_type)
completed = []
for stage in stages_to_check:
cp = read_checkpoint(pipeline_dir, project_id, stage)
if cp and cp.get("status") == "completed":
completed.append(stage)
return completed
def get_next_stage(
pipeline_dir: Path, project_id: str, pipeline_type: str | None = None
) -> Optional[str]:
"""Determine the next stage to run based on completed checkpoints.
Uses pipeline-specific stage order so that pipelines with different
stage sequences (e.g. cinematic vs explainer) progress correctly.
"""
stages = get_pipeline_stages(pipeline_type) if pipeline_type else STAGES
completed = set(get_completed_stages(pipeline_dir, project_id, pipeline_type))
for stage in stages:
if stage not in completed:
return stage
return None