How OpenMontage Uses JSON Schemas for Contract Validation Across Pipeline Stages

OpenMontage enforces strict data contracts across its video-editing pipeline by validating every stage's input and output against JSON Schema definitions stored in schemas/tools/, using the jsonschema library (>=4.20) in lib/pipeline_loader.py to halt execution immediately on any validation mismatch.

OpenMontage leverages JSON schemas for contract validation to guarantee type safety and structural integrity throughout its video-editing workflow. The calesthio/OpenMontage repository implements a schema-first architecture where each pipeline tool declares its data contracts in standalone .schema.json files, ensuring that malformed data never reaches downstream stages.

Schema Architecture and File Organization

All JSON Schema contracts reside in the schemas/tools/ directory at the repository root. Each tool defines its input and output structure using separate .schema.json files that explicitly declare allowed fields, types, required keys, and enum constraints.

Key schema definitions include:

• schemas/tools/video_stitch.schema.json – Defines input/output contracts for video stitching operations, specifying required parameters like file paths and duration limits
• schemas/tools/threejs_world.schema.json – Validates configuration objects for 3-D world generation, ensuring camera coordinates and scene parameters meet expected formats

The framework locates schemas dynamically using a naming convention where the tool name maps directly to its schema file, enabling the pipeline loader to resolve contracts without hardcoded paths.

How JSON Schemas Enable Contract Validation Across Pipeline Stages

The validation system operates across three critical components that ensure end-to-end contract safety:

Pipeline Loader (lib/pipeline_loader.py): Reads the user-provided playbook manifest and validates stage payloads using jsonschema.validate(instance=manifest, schema=schema). This component checks both the manifest structure and individual stage inputs before execution begins.

Playbook Generator (lib/playbook_generator.py): Produces JSON pipeline manifests and validates the generated output against a master playbook schema. It calls jsonschema.validate(instance=playbook, schema=schema) to ensure serializations conform to structural requirements before writing to disk.

Checkpoint Manager (lib/checkpoint.py): Persists intermediate pipeline results with schema validation via _load_checkpoint_schema(). This ensures that resumable checkpoints maintain data integrity across process restarts.

All components depend on the jsonschema library (version >=4.20) as declared in setup.py.

Step-by-Step Validation Implementation

Loading Schema Definitions

The framework loads schemas dynamically using filesystem resolution relative to the library location:

import json
import pathlib

def _load_schema(name: str) -> dict:
    schema_path = pathlib.Path(__file__).parent.parent / "schemas" / "tools" / f"{name}.schema.json"
    return json.loads(schema_path.read_text())

This utility constructs absolute paths to ensure schemas load correctly regardless of the execution context or working directory.

Runtime Validation Logic

Before executing any tool, the pipeline validates incoming payloads against the tool-specific contract:

import jsonschema

def _validate_payload(payload: dict, schema_name: str):
    schema = _load_schema(schema_name)
    jsonschema.validate(instance=payload, schema=schema)  # Raises ValidationError on mismatch

The orchestrator implements bidirectional validation, checking both inputs and outputs:

from .tools.video_stitch import stitch_videos

def run_video_stitch_stage(payload: dict) -> dict:
    _validate_payload(payload, "video_stitch")      # Input contract enforcement

    result = stitch_videos(**payload)               # Business logic execution

    _validate_payload(result, "video_stitch")       # Output contract verification

    return result

Error Handling and Pipeline Halting

When jsonschema.validate encounters a schema violation, it raises jsonschema.ValidationError. The pipeline loader catches this exception and terminates execution immediately, surfacing the specific validation failure to the user. This fail-fast behavior prevents malformed data from propagating through subsequent stages or corrupting final video outputs.

Practical Implementation Example

Below is a minimal reproducible example demonstrating manual contract validation as implemented in OpenMontage:

import jsonschema

# Schema definition (typically stored in schemas/tools/clip.schema.json)

clip_schema = {
    "type": "object",
    "properties": {
        "title": {"type": "string"},
        "duration": {"type": "number", "minimum": 0}
    },
    "required": ["title", "duration"],
    "additionalProperties": False
}

# Valid payload for a video clip stage

payload = {"title": "Intro Clip", "duration": 12.5}

# Validation execution (raises ValidationError if contract violated)

jsonschema.validate(instance=payload, schema=clip_schema)
print("✅ Payload conforms to JSON schema contract")

If the payload omitted the duration field or supplied a negative number, the validator would raise ValidationError with a detailed message indicating the specific schema constraint violation.

Contract Testing Strategy

The repository includes comprehensive test suites in tests/contracts/ that verify schema enforcement behavior:

• tests/contracts/test_taste_governance_contracts.py – Validates "taste governance" specific contracts, exercising both conforming and non-conforming payloads to ensure schemas correctly gate data flow
• tests/contracts/test_jimeng_video.py – Tests video-related schema validation, verifying that the pipeline rejects malformed video configurations while accepting valid specifications

These tests instantiate actual payloads and assert that jsonschema.ValidationError raises when contracts are violated, providing regression protection for the validation system.

Summary

• OpenMontage stores all JSON Schema contracts in schemas/tools/ with a .schema.json naming convention for each pipeline tool
• The pipeline loader (lib/pipeline_loader.py) validates every stage's input and output using jsonschema.validate before and after execution
• Bidirectional validation ensures both incoming payloads and tool results conform to defined contracts, preventing data corruption
• The system implements fail-fast error handling, raising jsonschema.ValidationError immediately upon mismatch to halt the pipeline
• Comprehensive contract tests in tests/contracts/test_taste_governance_contracts.py and tests/contracts/test_jimeng_video.py verify schema correctness

Frequently Asked Questions

Where are JSON schema files stored in OpenMontage?

JSON Schema files reside in the schemas/tools/ directory. Each tool uses a dedicated .schema.json file (such as video_stitch.schema.json or threejs_world.schema.json) that defines the valid structure for that stage's inputs and outputs.

What happens when a payload fails contract validation?

When validation fails, jsonschema.validate raises a jsonschema.ValidationError exception. The pipeline loader catches this error and immediately halts execution, preventing the invalid data from reaching the tool or subsequent stages. The error message includes specific details about which schema constraint was violated.

How does the playbook generator use JSON schemas?

The playbook generator (lib/playbook_generator.py) validates generated manifests against a master playbook schema before serialization. It calls jsonschema.validate(instance=playbook, schema=schema) to ensure the pipeline configuration conforms to expected structures before writing to disk.

Which Python library handles the validation?

OpenMontage uses the jsonschema library (version 4.20 or higher) to perform all contract validation. This library is declared as a dependency in setup.py and provides the validate() function that checks data instances against JSON Schema definitions.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}