feat(validator): schema-version dispatch + migrate-template.py framework (#18)

Closes the schema-versioning workstream of #90. Sets up the machinery
for "we will be updating a lot" (the user's framing) without forcing
the first real schema bump to discover semantics under deadline
pressure. Today every template is at v1; this PR adds the framework,
ships zero behavior change for v1 templates, and reserves v2+ for
when there's a concrete reason to bump.

Validator changes:

  - `KNOWN_SCHEMA_VERSIONS = {1}` — the set the validator currently
    accepts. Future bumps add to this set.
  - `DEPRECATED_SCHEMA_VERSIONS: set[int] = set()` — versions accepted
    with warning during a deprecation window.
  - Per-version contract: `_check_schema_v1(config)` enforces the v1
    REQUIRED_KEYS / OPTIONAL_KEYS / KNOWN_RUNTIMES contract — exactly
    what the previous monolithic check_config_yaml did.
  - Dispatch table: `SCHEMA_CHECKS = {1: _check_schema_v1}`. Versions
    that aren't in the table hard-error.

  - check_config_yaml() now: reads template_schema_version → emits
    deprecation warning if applicable → dispatches to the right
    SCHEMA_CHECKS entry → unknown versions hard-error with actionable
    instructions ("add a SCHEMA_V<N> block").

  - Schema versions are FROZEN once shipped: never edit a SCHEMA_V<N>
    constant in place. To bump, ADD v<N+1> alongside, deprecate v<N>,
    migrate consumers, drop v<N> next cycle. Header comment documents
    the discipline.

New script `migrate-template.py`:

  - `MIGRATIONS: dict[int, Callable[[dict], dict]]` registry — each
    entry maps a SOURCE version to the function that produces the
    next version's dict. Empty today.
  - `migrate_config(config, from, to)` chains migrations sequentially.
    Forward-only (errors on backward), errors on missing intermediate
    steps (never silently skip), asserts every migration stamps its
    output's template_schema_version.
  - CLI: `migrate-template.py [--from N] [--to M] [--dry-run] DIR`.
    Defaults: --from = whatever config.yaml declares, --to = highest
    reachable from MIGRATIONS (currently 1, so a no-op).

Behavior change to the existing
test_missing_required_keys_errors test:

  Previously the validator emitted 3 "missing required key" errors
  when name/runtime/template_schema_version were all missing. Now it
  short-circuits on missing version with a single actionable error —
  listing downstream missing keys is noise on top of the real
  problem (no version means we can't pick a contract). The test was
  updated to pin the new behavior; a new sibling test
  (test_missing_required_keys_under_v1_dispatch_errors) pins that v1
  still lists name/runtime/etc. when present-with-v1.

Verification:

  - 42/42 tests pass (20 prior + 9 new schema-dispatch tests in
    test_validate_workspace_template.py + 17 new migrator tests in
    test_migrate_template.py).
  - Real langgraph template runs through the full updated validator
    end-to-end with 0 warnings / 0 errors.

This + #17 means #90 is done end-to-end:
  - Phase 2: validator green on all 8 templates as a required check (already shipped)
  - Phase 2.5: adapter.py runtime-load contract (#17)
  - Phase 3: schema versioning + migration framework (this PR)

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Hongming Wang 2026-04-28 12:07:04 -07:00 committed by GitHub
parent 8309a55e6c
commit 84a104a146
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 947 additions and 35 deletions

View File

@ -0,0 +1,212 @@
#!/usr/bin/env python3
"""Migrate a workspace template's config.yaml across schema versions.
Companion to validate-workspace-template.py. Whenever the validator
adds a new schema version, this script gets a corresponding entry in
MIGRATIONS so each consumer template can mechanically upgrade rather
than every maintainer figuring out the field changes by hand.
Discipline (matches the validator's header):
1. Validator gets a SCHEMA_V<N+1> block + KNOWN_SCHEMA_VERSIONS bump.
2. This script gets `MIGRATIONS[N]` defined a function that takes
a v<N> dict and returns a v<N+1> dict. Pure, deterministic, no
I/O that way migrations compose: v1 v2 v3 just chains them.
3. Each migration is FROZEN once shipped. If a v2 migration needs
fixing post-ship, ship it as v3 with the corrective migration.
4. Consumers run this script (one PR per template repo) before the
deprecation window for v<N> closes.
Usage:
# Migrate the template in cwd from its current version to the latest
python3 scripts/migrate-template.py .
# Migrate to a specific version (bounded; useful when a deprecation
# window is closing and you want to skip-ahead)
python3 scripts/migrate-template.py --to 3 .
# Force the source version (override config.yaml's declared version)
python3 scripts/migrate-template.py --from 1 --to 2 .
# Dry-run: print the diff without writing
python3 scripts/migrate-template.py --dry-run .
The script preserves YAML round-trip fidelity for keys it doesn't
touch (using ruamel.yaml when available; falling back to PyYAML's
default representer otherwise). Migrations should ONLY mutate keys
they're explicitly versioning — leave everything else alone so a
consumer template's customizations survive.
"""
from __future__ import annotations
import argparse
import sys
from copy import deepcopy
from pathlib import Path
from typing import Callable
import yaml
# ──────────────────────────────────────────── migrations registry
# Each entry maps a SOURCE version to the function that produces the
# next version's dict. Currently empty — no v2 yet. The first time a
# real schema bump lands, MIGRATIONS[1] gets defined alongside the
# validator's SCHEMA_V2 block.
MIGRATIONS: dict[int, Callable[[dict], dict]] = {}
# ──────────────────────────────────────────── version detection
def _detect_current_version(config: dict) -> int:
sv = config.get("template_schema_version")
if sv is None:
sys.exit(
"error: config.yaml has no `template_schema_version`. "
"Add it (likely 1 for legacy templates) before migrating."
)
if not isinstance(sv, int):
sys.exit(
f"error: template_schema_version must be int, got "
f"{type(sv).__name__}={sv!r}."
)
return sv
def _latest_known_version() -> int:
"""Maximum version reachable by chaining MIGRATIONS from any
starting point. With an empty registry, this is 1 (the floor:
every existing template is at v1)."""
if not MIGRATIONS:
return 1
return max(MIGRATIONS.keys()) + 1
# ──────────────────────────────────────────── core
def migrate_config(config: dict, from_version: int, to_version: int) -> dict:
"""Apply migrations sequentially from `from_version` to `to_version`.
Returns a NEW dict does not mutate the input.
Errors loudly when there's no migration registered for an
intermediate step: forward-only, never silently skip a hop. If the
user asks for a backward migration, error too schema versions
are append-only and we don't ship downgrades."""
if to_version < from_version:
sys.exit(
f"error: cannot migrate backward (from v{from_version} to "
f"v{to_version}). Schema versions are append-only — file a "
f"new bug + ship a forward migration instead."
)
current = from_version
out = deepcopy(config)
while current < to_version:
step = MIGRATIONS.get(current)
if step is None:
sys.exit(
f"error: no migration registered for v{current}"
f"v{current + 1}. Either add it to MIGRATIONS in "
f"scripts/migrate-template.py or pick a different --to."
)
out = step(out)
# Every migration MUST stamp the new version on its output —
# this assertion catches a class of bugs where a migration
# forgets to bump template_schema_version.
if out.get("template_schema_version") != current + 1:
sys.exit(
f"error: MIGRATIONS[{current}] did not stamp "
f"template_schema_version={current + 1} on its output. "
f"This is a bug in the migration function itself."
)
current += 1
return out
def _read_yaml(path: Path) -> dict:
with open(path) as f:
data = yaml.safe_load(f)
if not isinstance(data, dict):
sys.exit(f"error: {path} root is not a mapping (got {type(data).__name__})")
return data
def _write_yaml(path: Path, data: dict) -> None:
# Sort keys for stable diffs across migrations. This matches what
# `yaml.safe_dump` does when we write — consumer repos with
# custom orderings will see their config.yaml re-ordered, which is
# one of those round-trip lossy tradeoffs that's worth accepting:
# the migration moment is rare and the diff is reviewable.
with open(path, "w") as f:
yaml.safe_dump(data, f, sort_keys=True, default_flow_style=False)
# ──────────────────────────────────────────── CLI
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(
description="Migrate a workspace template's config.yaml across schema versions."
)
parser.add_argument(
"template_dir",
type=Path,
help="Path to the template repo root (must contain config.yaml).",
)
parser.add_argument(
"--from",
dest="from_version",
type=int,
default=None,
help="Source schema version (defaults to whatever config.yaml declares).",
)
parser.add_argument(
"--to",
dest="to_version",
type=int,
default=None,
help="Target schema version (defaults to the highest reachable from MIGRATIONS).",
)
parser.add_argument(
"--dry-run",
action="store_true",
help="Print the migrated YAML to stdout without modifying the file.",
)
args = parser.parse_args(argv)
config_path = args.template_dir / "config.yaml"
if not config_path.is_file():
sys.exit(f"error: {config_path} does not exist")
config = _read_yaml(config_path)
from_version = args.from_version
if from_version is None:
from_version = _detect_current_version(config)
to_version = args.to_version
if to_version is None:
to_version = _latest_known_version()
if from_version == to_version:
print(
f"nothing to do: config.yaml is already at v{from_version}.",
file=sys.stderr,
)
return 0
migrated = migrate_config(config, from_version, to_version)
if args.dry_run:
yaml.safe_dump(migrated, sys.stdout, sort_keys=True, default_flow_style=False)
return 0
_write_yaml(config_path, migrated)
print(
f"✓ migrated {config_path} from v{from_version} → v{to_version}",
file=sys.stderr,
)
return 0
if __name__ == "__main__":
sys.exit(main())

View File

@ -101,8 +101,36 @@ KNOWN_RUNTIMES = {
"gemini-cli", "gemini-cli",
"openclaw", "openclaw",
} }
REQUIRED_KEYS = ["name", "runtime", "template_schema_version"]
OPTIONAL_KEYS = [ # ──────────────────────────────────────────── schema versioning
#
# `template_schema_version: int` in each template's config.yaml selects
# which contract this validator enforces. Versions are FROZEN once
# shipped — never edit a SCHEMA_V* constant in place. To bump:
#
# 1. Add `SCHEMA_V<N+1>_REQUIRED_KEYS` / `SCHEMA_V<N+1>_OPTIONAL_KEYS`
# describing the new contract.
# 2. Add `_check_schema_v<N+1>(config)` that enforces it.
# 3. Add the entry to SCHEMA_CHECKS below.
# 4. Move version N from KNOWN_SCHEMA_VERSIONS to
# DEPRECATED_SCHEMA_VERSIONS so existing v<N> templates warn but
# still pass — buys a deprecation window.
# 5. Ship a corresponding migration in scripts/migrate-template.py's
# MIGRATIONS table (key = N, value = callable that produces the
# v<N+1> dict from a v<N> dict).
# 6. Run migrate-template.py on each consumer template repo as a PR.
# 7. After all consumers migrate, drop version N from
# DEPRECATED_SCHEMA_VERSIONS in a follow-up PR.
#
# This discipline means a schema version always has exactly one valid
# enforcement function, never "branch on minor variants" — the whole
# point of versioning is to avoid that drift.
KNOWN_SCHEMA_VERSIONS: set[int] = {1}
DEPRECATED_SCHEMA_VERSIONS: set[int] = set()
SCHEMA_V1_REQUIRED_KEYS = ["name", "runtime", "template_schema_version"]
SCHEMA_V1_OPTIONAL_KEYS = [
"description", "description",
"version", "version",
"tier", "tier",
@ -120,6 +148,33 @@ OPTIONAL_KEYS = [
] ]
def _check_schema_v1(config: dict) -> None:
"""v1 contract — the keys frozen as of monorepo task #90's Phase 2.
Currently every production template runs this version. Do NOT edit
in place; add v2 instead and migrate consumers (see header)."""
for key in SCHEMA_V1_REQUIRED_KEYS:
if key not in config:
err(f"config.yaml: missing required key `{key}`")
runtime = config.get("runtime")
if runtime and runtime not in KNOWN_RUNTIMES:
warn(
f"config.yaml: runtime `{runtime}` not in known set "
f"{sorted(KNOWN_RUNTIMES)} — OK for custom runtimes; "
f"if canonical, add it to KNOWN_RUNTIMES in validate-workspace-template.py"
)
unknown = set(config.keys()) - set(SCHEMA_V1_REQUIRED_KEYS) - set(SCHEMA_V1_OPTIONAL_KEYS)
if unknown:
warn(
f"config.yaml: unknown top-level keys {sorted(unknown)}"
f"may be drift. If intentional, add them to SCHEMA_V1_OPTIONAL_KEYS."
)
SCHEMA_CHECKS = {
1: _check_schema_v1,
}
def check_config_yaml() -> None: def check_config_yaml() -> None:
if not os.path.isfile("config.yaml"): if not os.path.isfile("config.yaml"):
err("config.yaml: missing at repo root") err("config.yaml: missing at repo root")
@ -133,29 +188,40 @@ def check_config_yaml() -> None:
if not isinstance(config, dict): if not isinstance(config, dict):
err(f"config.yaml: root must be a mapping, got {type(config).__name__}") err(f"config.yaml: root must be a mapping, got {type(config).__name__}")
return return
for key in REQUIRED_KEYS:
if key not in config: # Schema-version dispatch. Validate the version field shape first
err(f"config.yaml: missing required key `{key}`") # so error messages are actionable.
runtime = config.get("runtime")
if runtime and runtime not in KNOWN_RUNTIMES:
warn(
f"config.yaml: runtime `{runtime}` not in known set "
f"{sorted(KNOWN_RUNTIMES)} — OK for custom runtimes; "
f"if canonical, add it to KNOWN_RUNTIMES in validate-workspace-template.py"
)
sv = config.get("template_schema_version") sv = config.get("template_schema_version")
if sv is not None and not isinstance(sv, int): if sv is None:
err("config.yaml: missing required key `template_schema_version`")
# Can't dispatch without a version. Don't fall through to v1
# checks — that would mask the missing-version error.
return
if not isinstance(sv, int):
err( err(
f"config.yaml: template_schema_version must be int, " f"config.yaml: template_schema_version must be int, "
f"got {type(sv).__name__}={sv!r}" f"got {type(sv).__name__}={sv!r}"
) )
return
unknown = set(config.keys()) - set(REQUIRED_KEYS) - set(OPTIONAL_KEYS) if sv in DEPRECATED_SCHEMA_VERSIONS:
if unknown: latest = max(KNOWN_SCHEMA_VERSIONS)
warn( warn(
f"config.yaml: unknown top-level keys {sorted(unknown)}" f"config.yaml: template_schema_version={sv} is deprecated; "
f"may be drift. If intentional, add them to OPTIONAL_KEYS." f"migrate to v{latest} via "
f"`python3 scripts/migrate-template.py --to {latest} .`. "
f"Support for v{sv} will be removed in a future cycle."
) )
elif sv not in KNOWN_SCHEMA_VERSIONS:
valid = sorted(KNOWN_SCHEMA_VERSIONS | DEPRECATED_SCHEMA_VERSIONS)
err(
f"config.yaml: template_schema_version={sv} is unknown — "
f"this validator understands {valid}. Either bump the "
f"validator (add a SCHEMA_V{sv} block) or correct the version."
)
return
SCHEMA_CHECKS[sv](config)
# ───────────────────────────────────────────────────────────── requirements.txt # ───────────────────────────────────────────────────────────── requirements.txt

212
scripts/migrate-template.py Executable file
View File

@ -0,0 +1,212 @@
#!/usr/bin/env python3
"""Migrate a workspace template's config.yaml across schema versions.
Companion to validate-workspace-template.py. Whenever the validator
adds a new schema version, this script gets a corresponding entry in
MIGRATIONS so each consumer template can mechanically upgrade rather
than every maintainer figuring out the field changes by hand.
Discipline (matches the validator's header):
1. Validator gets a SCHEMA_V<N+1> block + KNOWN_SCHEMA_VERSIONS bump.
2. This script gets `MIGRATIONS[N]` defined a function that takes
a v<N> dict and returns a v<N+1> dict. Pure, deterministic, no
I/O that way migrations compose: v1 v2 v3 just chains them.
3. Each migration is FROZEN once shipped. If a v2 migration needs
fixing post-ship, ship it as v3 with the corrective migration.
4. Consumers run this script (one PR per template repo) before the
deprecation window for v<N> closes.
Usage:
# Migrate the template in cwd from its current version to the latest
python3 scripts/migrate-template.py .
# Migrate to a specific version (bounded; useful when a deprecation
# window is closing and you want to skip-ahead)
python3 scripts/migrate-template.py --to 3 .
# Force the source version (override config.yaml's declared version)
python3 scripts/migrate-template.py --from 1 --to 2 .
# Dry-run: print the diff without writing
python3 scripts/migrate-template.py --dry-run .
The script preserves YAML round-trip fidelity for keys it doesn't
touch (using ruamel.yaml when available; falling back to PyYAML's
default representer otherwise). Migrations should ONLY mutate keys
they're explicitly versioning — leave everything else alone so a
consumer template's customizations survive.
"""
from __future__ import annotations
import argparse
import sys
from copy import deepcopy
from pathlib import Path
from typing import Callable
import yaml
# ──────────────────────────────────────────── migrations registry
# Each entry maps a SOURCE version to the function that produces the
# next version's dict. Currently empty — no v2 yet. The first time a
# real schema bump lands, MIGRATIONS[1] gets defined alongside the
# validator's SCHEMA_V2 block.
MIGRATIONS: dict[int, Callable[[dict], dict]] = {}
# ──────────────────────────────────────────── version detection
def _detect_current_version(config: dict) -> int:
sv = config.get("template_schema_version")
if sv is None:
sys.exit(
"error: config.yaml has no `template_schema_version`. "
"Add it (likely 1 for legacy templates) before migrating."
)
if not isinstance(sv, int):
sys.exit(
f"error: template_schema_version must be int, got "
f"{type(sv).__name__}={sv!r}."
)
return sv
def _latest_known_version() -> int:
"""Maximum version reachable by chaining MIGRATIONS from any
starting point. With an empty registry, this is 1 (the floor:
every existing template is at v1)."""
if not MIGRATIONS:
return 1
return max(MIGRATIONS.keys()) + 1
# ──────────────────────────────────────────── core
def migrate_config(config: dict, from_version: int, to_version: int) -> dict:
"""Apply migrations sequentially from `from_version` to `to_version`.
Returns a NEW dict does not mutate the input.
Errors loudly when there's no migration registered for an
intermediate step: forward-only, never silently skip a hop. If the
user asks for a backward migration, error too schema versions
are append-only and we don't ship downgrades."""
if to_version < from_version:
sys.exit(
f"error: cannot migrate backward (from v{from_version} to "
f"v{to_version}). Schema versions are append-only — file a "
f"new bug + ship a forward migration instead."
)
current = from_version
out = deepcopy(config)
while current < to_version:
step = MIGRATIONS.get(current)
if step is None:
sys.exit(
f"error: no migration registered for v{current}"
f"v{current + 1}. Either add it to MIGRATIONS in "
f"scripts/migrate-template.py or pick a different --to."
)
out = step(out)
# Every migration MUST stamp the new version on its output —
# this assertion catches a class of bugs where a migration
# forgets to bump template_schema_version.
if out.get("template_schema_version") != current + 1:
sys.exit(
f"error: MIGRATIONS[{current}] did not stamp "
f"template_schema_version={current + 1} on its output. "
f"This is a bug in the migration function itself."
)
current += 1
return out
def _read_yaml(path: Path) -> dict:
with open(path) as f:
data = yaml.safe_load(f)
if not isinstance(data, dict):
sys.exit(f"error: {path} root is not a mapping (got {type(data).__name__})")
return data
def _write_yaml(path: Path, data: dict) -> None:
# Sort keys for stable diffs across migrations. This matches what
# `yaml.safe_dump` does when we write — consumer repos with
# custom orderings will see their config.yaml re-ordered, which is
# one of those round-trip lossy tradeoffs that's worth accepting:
# the migration moment is rare and the diff is reviewable.
with open(path, "w") as f:
yaml.safe_dump(data, f, sort_keys=True, default_flow_style=False)
# ──────────────────────────────────────────── CLI
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(
description="Migrate a workspace template's config.yaml across schema versions."
)
parser.add_argument(
"template_dir",
type=Path,
help="Path to the template repo root (must contain config.yaml).",
)
parser.add_argument(
"--from",
dest="from_version",
type=int,
default=None,
help="Source schema version (defaults to whatever config.yaml declares).",
)
parser.add_argument(
"--to",
dest="to_version",
type=int,
default=None,
help="Target schema version (defaults to the highest reachable from MIGRATIONS).",
)
parser.add_argument(
"--dry-run",
action="store_true",
help="Print the migrated YAML to stdout without modifying the file.",
)
args = parser.parse_args(argv)
config_path = args.template_dir / "config.yaml"
if not config_path.is_file():
sys.exit(f"error: {config_path} does not exist")
config = _read_yaml(config_path)
from_version = args.from_version
if from_version is None:
from_version = _detect_current_version(config)
to_version = args.to_version
if to_version is None:
to_version = _latest_known_version()
if from_version == to_version:
print(
f"nothing to do: config.yaml is already at v{from_version}.",
file=sys.stderr,
)
return 0
migrated = migrate_config(config, from_version, to_version)
if args.dry_run:
yaml.safe_dump(migrated, sys.stdout, sort_keys=True, default_flow_style=False)
return 0
_write_yaml(config_path, migrated)
print(
f"✓ migrated {config_path} from v{from_version} → v{to_version}",
file=sys.stderr,
)
return 0
if __name__ == "__main__":
sys.exit(main())

View File

@ -0,0 +1,242 @@
"""Tests for migrate-template.py — pin the migration framework's
behavior so the FIRST real schema bump (the one that proves the system
end-to-end) doesn't have to discover semantics under deadline pressure.
The MIGRATIONS registry is empty today (we have only v1), so most
tests register a synthetic migration scoped to the test, exercise the
machinery, and unregister at teardown. This way the framework's
contract is locked in even before any real migration ships.
"""
from __future__ import annotations
import importlib.util
from pathlib import Path
import pytest
MIGRATOR_PATH = Path(__file__).resolve().parent / "migrate-template.py"
def _load_migrator():
"""Load migrate-template.py by path (its filename has a hyphen so
we can't `import migrate-template` directly)."""
spec = importlib.util.spec_from_file_location("migrator", MIGRATOR_PATH)
assert spec is not None and spec.loader is not None
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)
return mod
@pytest.fixture
def migrator():
"""Fresh migrator module per test. Registry is global module
state; tests that register synthetic migrations must clean up."""
mod = _load_migrator()
# Snapshot + restore MIGRATIONS so accidentally-leaked entries
# from one test don't poison the next.
snapshot = dict(mod.MIGRATIONS)
yield mod
mod.MIGRATIONS.clear()
mod.MIGRATIONS.update(snapshot)
def _v1_template_config() -> dict:
return {
"name": "test-template",
"runtime": "claude-code",
"template_schema_version": 1,
"description": "fixture",
"tier": 1,
}
# ─────────────────────────────────────── version detection
def test_detect_current_version_from_config(migrator):
config = _v1_template_config()
assert migrator._detect_current_version(config) == 1
def test_detect_missing_version_exits(migrator):
config = {"name": "t", "runtime": "claude-code"}
with pytest.raises(SystemExit) as exc:
migrator._detect_current_version(config)
assert "no `template_schema_version`" in str(exc.value)
def test_detect_non_int_version_exits(migrator):
config = {"name": "t", "runtime": "claude-code", "template_schema_version": "1"}
with pytest.raises(SystemExit) as exc:
migrator._detect_current_version(config)
assert "must be int" in str(exc.value)
# ─────────────────────────────────────── latest-version reachability
def test_latest_with_empty_registry_is_v1(migrator):
"""Floor case: every existing template is v1 even when no
migrations are registered. Latest reachable = v1, so a no-op
migration is the only valid action."""
migrator.MIGRATIONS.clear()
assert migrator._latest_known_version() == 1
def test_latest_with_one_migration_is_v2(migrator):
"""Adding a v1 → v2 migration moves the ceiling to v2. This is
what happens the first time a real schema bump ships."""
migrator.MIGRATIONS.clear()
migrator.MIGRATIONS[1] = lambda c: {**c, "template_schema_version": 2}
assert migrator._latest_known_version() == 2
def test_latest_chains_through_multiple_migrations(migrator):
"""Multi-step ceiling: v1 → v2 → v3 chain produces ceiling=3."""
migrator.MIGRATIONS.clear()
migrator.MIGRATIONS[1] = lambda c: {**c, "template_schema_version": 2}
migrator.MIGRATIONS[2] = lambda c: {**c, "template_schema_version": 3}
assert migrator._latest_known_version() == 3
# ─────────────────────────────────────── migrate_config core
def test_migrate_no_op_when_versions_match(migrator):
"""from == to → no migration step runs. Should not require any
MIGRATIONS entry to be defined."""
migrator.MIGRATIONS.clear()
out = migrator.migrate_config(_v1_template_config(), 1, 1)
assert out == _v1_template_config()
assert out is not _v1_template_config() # deep-copied, not aliased
def test_migrate_one_step_applies_function(migrator):
"""v1 → v2 with a registered migration produces the expected
output and stamps the new version."""
migrator.MIGRATIONS.clear()
migrator.MIGRATIONS[1] = lambda c: {**c, "template_schema_version": 2, "added_in_v2": True}
out = migrator.migrate_config(_v1_template_config(), 1, 2)
assert out["template_schema_version"] == 2
assert out["added_in_v2"] is True
# Pre-existing keys preserved.
assert out["name"] == "test-template"
def test_migrate_chains_v1_to_v3(migrator):
"""Two-step migration: v1 → v2 → v3. Each step applies in order."""
migrator.MIGRATIONS.clear()
migrator.MIGRATIONS[1] = lambda c: {**c, "template_schema_version": 2, "from_v1": True}
migrator.MIGRATIONS[2] = lambda c: {**c, "template_schema_version": 3, "from_v2": True}
out = migrator.migrate_config(_v1_template_config(), 1, 3)
assert out["template_schema_version"] == 3
assert out["from_v1"] is True
assert out["from_v2"] is True
def test_migrate_missing_step_exits(migrator):
"""If MIGRATIONS lacks the v<current> step, fail loud rather than
silently skip the version. Forward-only, never silent skip."""
migrator.MIGRATIONS.clear()
# No MIGRATIONS[1] registered.
with pytest.raises(SystemExit) as exc:
migrator.migrate_config(_v1_template_config(), 1, 2)
assert "no migration registered for v1 → v2" in str(exc.value)
def test_migrate_backward_exits(migrator):
"""Schema versions are append-only. Asking for v2 → v1 must
error, not silently downgrade."""
migrator.MIGRATIONS.clear()
config = {**_v1_template_config(), "template_schema_version": 2}
with pytest.raises(SystemExit) as exc:
migrator.migrate_config(config, 2, 1)
assert "cannot migrate backward" in str(exc.value)
def test_migration_must_stamp_new_version(migrator):
"""A migration function that forgets to bump
`template_schema_version` is a bug catch it at apply time so
the framework can never produce an inconsistent output."""
migrator.MIGRATIONS.clear()
# Buggy migration: doesn't update the version field.
migrator.MIGRATIONS[1] = lambda c: {**c, "added_in_v2": True}
with pytest.raises(SystemExit) as exc:
migrator.migrate_config(_v1_template_config(), 1, 2)
assert "did not stamp template_schema_version=2" in str(exc.value)
def test_migrate_does_not_mutate_input(migrator):
"""migrate_config returns a fresh dict; the caller's input is
untouched. Pin this so a shared-state migration can't accidentally
poison the caller's view of the original template."""
migrator.MIGRATIONS.clear()
migrator.MIGRATIONS[1] = lambda c: {**c, "template_schema_version": 2}
original = _v1_template_config()
snapshot = dict(original)
_ = migrator.migrate_config(original, 1, 2)
assert original == snapshot
# ─────────────────────────────────────── CLI smoke
def test_cli_writes_migrated_yaml(migrator, tmp_path):
"""End-to-end: --to migrates the file in place and exits 0."""
migrator.MIGRATIONS.clear()
migrator.MIGRATIONS[1] = lambda c: {**c, "template_schema_version": 2, "added": "v2-marker"}
cfg = tmp_path / "config.yaml"
cfg.write_text(
"name: t\n"
"runtime: claude-code\n"
"template_schema_version: 1\n"
)
rc = migrator.main([str(tmp_path), "--to", "2"])
assert rc == 0
written = cfg.read_text()
assert "template_schema_version: 2" in written
assert "added: v2-marker" in written
def test_cli_dry_run_does_not_modify_file(migrator, tmp_path, capsys):
"""--dry-run prints the migrated YAML to stdout but leaves the
on-disk file untouched."""
migrator.MIGRATIONS.clear()
migrator.MIGRATIONS[1] = lambda c: {**c, "template_schema_version": 2}
cfg = tmp_path / "config.yaml"
cfg.write_text(
"name: t\n"
"runtime: claude-code\n"
"template_schema_version: 1\n"
)
original_disk = cfg.read_text()
rc = migrator.main([str(tmp_path), "--to", "2", "--dry-run"])
assert rc == 0
assert cfg.read_text() == original_disk # untouched
captured = capsys.readouterr()
assert "template_schema_version: 2" in captured.out
def test_cli_no_op_when_already_at_target(migrator, tmp_path, capsys):
"""If the template is already at the target version, exit 0
without modifying the file. Not an error common when running
the migration script defensively in CI."""
migrator.MIGRATIONS.clear()
cfg = tmp_path / "config.yaml"
cfg.write_text(
"name: t\n"
"runtime: claude-code\n"
"template_schema_version: 1\n"
)
original = cfg.read_text()
rc = migrator.main([str(tmp_path), "--to", "1"])
assert rc == 0
assert cfg.read_text() == original
def test_cli_missing_config_exits(migrator, tmp_path):
"""If the target dir has no config.yaml, error clearly rather
than try to apply migrations to nothing."""
with pytest.raises(SystemExit) as exc:
migrator.main([str(tmp_path), "--to", "2"])
assert "config.yaml" in str(exc.value) and "does not exist" in str(exc.value)

View File

@ -200,13 +200,40 @@ def test_missing_entrypoint_errors(validator, tmp_path, monkeypatch):
# ───────────────────────────────────────────────────────── config.yaml drift # ───────────────────────────────────────────────────────── config.yaml drift
def test_missing_required_keys_errors(validator, tmp_path, monkeypatch): def test_missing_required_keys_errors(validator, tmp_path, monkeypatch):
"""A config without template_schema_version short-circuits with a
SINGLE actionable error listing 'also name and runtime are
missing' is noise on top of the real problem (no version means the
validator can't pick a schema contract to enforce). Once the
version is present, the v1 dispatch will list the other missing
keys (next test pins that)."""
cfg = "description: only description, no name/runtime/version\n" cfg = "description: only description, no name/runtime/version\n"
_materialise(tmp_path, dockerfile=_good_dockerfile(), config_yaml=cfg, _materialise(tmp_path, dockerfile=_good_dockerfile(), config_yaml=cfg,
requirements=_good_requirements_txt()) requirements=_good_requirements_txt())
monkeypatch.chdir(tmp_path) monkeypatch.chdir(tmp_path)
validator.check_config_yaml() validator.check_config_yaml()
missing_msgs = [e for e in validator.ERRORS if "missing required key" in e] missing_msgs = [e for e in validator.ERRORS if "missing required key" in e]
assert len(missing_msgs) >= 3 # name, runtime, template_schema_version # Exactly one error: the missing version. v1 dispatch is skipped
# because we can't choose a contract without a version.
assert len(missing_msgs) == 1, missing_msgs
assert "template_schema_version" in missing_msgs[0]
def test_missing_required_keys_under_v1_dispatch_errors(validator, tmp_path, monkeypatch):
"""When `template_schema_version: 1` IS present but other required
keys are missing, the v1 dispatch fires and lists them. Pins that
the v1 contract still enforces name + runtime."""
cfg = (
"template_schema_version: 1\n"
"description: only the version + description\n"
)
_materialise(tmp_path, dockerfile=_good_dockerfile(), config_yaml=cfg,
requirements=_good_requirements_txt())
monkeypatch.chdir(tmp_path)
validator.check_config_yaml()
missing_msgs = [e for e in validator.ERRORS if "missing required key" in e]
keys = {e.split("`")[1] for e in missing_msgs}
assert "name" in keys, missing_msgs
assert "runtime" in keys, missing_msgs
def test_string_template_schema_version_errors(validator, tmp_path, monkeypatch): def test_string_template_schema_version_errors(validator, tmp_path, monkeypatch):
@ -382,6 +409,93 @@ def test_adapter_with_import_error_errors(validator, tmp_path, monkeypatch):
), validator.ERRORS ), validator.ERRORS
# ─────────────────────────────────────── schema-version dispatch
#
# Pin the contract that the validator routes to per-version checks
# based on `template_schema_version`, that unknown versions hard-fail,
# and that deprecated versions warn but pass.
def test_v1_is_in_known_schema_versions(validator):
"""Document the floor: v1 is always understood. Future bumps add
versions; v1 stays accepted (or deprecated) but the validator
never silently drops it."""
assert 1 in validator.KNOWN_SCHEMA_VERSIONS or 1 in validator.DEPRECATED_SCHEMA_VERSIONS
def test_unknown_schema_version_errors(validator, tmp_path, monkeypatch):
"""A template declaring template_schema_version=999 must hard-fail
silently allowing it would let drift land disguised as a
'future' version."""
cfg = (
"name: t\n"
"runtime: claude-code\n"
"template_schema_version: 999\n"
)
_materialise(tmp_path, dockerfile=_good_dockerfile(), config_yaml=cfg,
requirements=_good_requirements_txt())
monkeypatch.chdir(tmp_path)
validator.check_config_yaml()
assert any("template_schema_version=999 is unknown" in e
for e in validator.ERRORS), validator.ERRORS
def test_deprecated_schema_version_warns_but_passes(validator, tmp_path, monkeypatch):
"""During a deprecation window, v<N-1> templates still validate
(so the consumer can keep merging unrelated PRs while migrating)
but the warning surfaces the migration command."""
# Inject a fake deprecated version for the duration of this test —
# we don't have a real deprecated version yet (only v1 exists).
validator.KNOWN_SCHEMA_VERSIONS.add(2)
validator.DEPRECATED_SCHEMA_VERSIONS.add(1)
validator.SCHEMA_CHECKS[2] = lambda config: None # accept-all stub for v2
try:
cfg = (
"name: t\n"
"runtime: claude-code\n"
"template_schema_version: 1\n"
)
_materialise(tmp_path, dockerfile=_good_dockerfile(), config_yaml=cfg,
requirements=_good_requirements_txt())
monkeypatch.chdir(tmp_path)
validator.check_config_yaml()
# No errors — deprecation is warning-only.
assert validator.ERRORS == [], validator.ERRORS
assert any(
"template_schema_version=1 is deprecated" in w
and "migrate-template.py" in w
for w in validator.WARNINGS
), validator.WARNINGS
finally:
validator.KNOWN_SCHEMA_VERSIONS.discard(2)
validator.DEPRECATED_SCHEMA_VERSIONS.discard(1)
validator.SCHEMA_CHECKS.pop(2, None)
def test_per_version_dispatch_calls_correct_check(validator, tmp_path, monkeypatch):
"""Pin that SCHEMA_CHECKS[N] is the function called when a template
declares template_schema_version=N. Without this, the dispatch could
fire the wrong contract on a multi-version codebase."""
called: list[int] = []
validator.KNOWN_SCHEMA_VERSIONS.add(7)
validator.SCHEMA_CHECKS[7] = lambda config: called.append(7)
try:
cfg = (
"name: t\n"
"runtime: claude-code\n"
"template_schema_version: 7\n"
)
_materialise(tmp_path, dockerfile=_good_dockerfile(), config_yaml=cfg,
requirements=_good_requirements_txt())
monkeypatch.chdir(tmp_path)
validator.check_config_yaml()
assert called == [7], f"v7 dispatch was not invoked; called={called}"
finally:
validator.KNOWN_SCHEMA_VERSIONS.discard(7)
validator.SCHEMA_CHECKS.pop(7, None)
def test_runtime_not_installed_warns_not_errors(validator, tmp_path, monkeypatch): def test_runtime_not_installed_warns_not_errors(validator, tmp_path, monkeypatch):
"""If the validator runs in an env without molecule-ai-workspace-runtime, """If the validator runs in an env without molecule-ai-workspace-runtime,
we WARN (loud) but don't error — hard-erroring would say 'your adapter we WARN (loud) but don't error — hard-erroring would say 'your adapter

View File

@ -101,8 +101,36 @@ KNOWN_RUNTIMES = {
"gemini-cli", "gemini-cli",
"openclaw", "openclaw",
} }
REQUIRED_KEYS = ["name", "runtime", "template_schema_version"]
OPTIONAL_KEYS = [ # ──────────────────────────────────────────── schema versioning
#
# `template_schema_version: int` in each template's config.yaml selects
# which contract this validator enforces. Versions are FROZEN once
# shipped — never edit a SCHEMA_V* constant in place. To bump:
#
# 1. Add `SCHEMA_V<N+1>_REQUIRED_KEYS` / `SCHEMA_V<N+1>_OPTIONAL_KEYS`
# describing the new contract.
# 2. Add `_check_schema_v<N+1>(config)` that enforces it.
# 3. Add the entry to SCHEMA_CHECKS below.
# 4. Move version N from KNOWN_SCHEMA_VERSIONS to
# DEPRECATED_SCHEMA_VERSIONS so existing v<N> templates warn but
# still pass — buys a deprecation window.
# 5. Ship a corresponding migration in scripts/migrate-template.py's
# MIGRATIONS table (key = N, value = callable that produces the
# v<N+1> dict from a v<N> dict).
# 6. Run migrate-template.py on each consumer template repo as a PR.
# 7. After all consumers migrate, drop version N from
# DEPRECATED_SCHEMA_VERSIONS in a follow-up PR.
#
# This discipline means a schema version always has exactly one valid
# enforcement function, never "branch on minor variants" — the whole
# point of versioning is to avoid that drift.
KNOWN_SCHEMA_VERSIONS: set[int] = {1}
DEPRECATED_SCHEMA_VERSIONS: set[int] = set()
SCHEMA_V1_REQUIRED_KEYS = ["name", "runtime", "template_schema_version"]
SCHEMA_V1_OPTIONAL_KEYS = [
"description", "description",
"version", "version",
"tier", "tier",
@ -120,6 +148,33 @@ OPTIONAL_KEYS = [
] ]
def _check_schema_v1(config: dict) -> None:
"""v1 contract — the keys frozen as of monorepo task #90's Phase 2.
Currently every production template runs this version. Do NOT edit
in place; add v2 instead and migrate consumers (see header)."""
for key in SCHEMA_V1_REQUIRED_KEYS:
if key not in config:
err(f"config.yaml: missing required key `{key}`")
runtime = config.get("runtime")
if runtime and runtime not in KNOWN_RUNTIMES:
warn(
f"config.yaml: runtime `{runtime}` not in known set "
f"{sorted(KNOWN_RUNTIMES)} — OK for custom runtimes; "
f"if canonical, add it to KNOWN_RUNTIMES in validate-workspace-template.py"
)
unknown = set(config.keys()) - set(SCHEMA_V1_REQUIRED_KEYS) - set(SCHEMA_V1_OPTIONAL_KEYS)
if unknown:
warn(
f"config.yaml: unknown top-level keys {sorted(unknown)}"
f"may be drift. If intentional, add them to SCHEMA_V1_OPTIONAL_KEYS."
)
SCHEMA_CHECKS = {
1: _check_schema_v1,
}
def check_config_yaml() -> None: def check_config_yaml() -> None:
if not os.path.isfile("config.yaml"): if not os.path.isfile("config.yaml"):
err("config.yaml: missing at repo root") err("config.yaml: missing at repo root")
@ -133,29 +188,40 @@ def check_config_yaml() -> None:
if not isinstance(config, dict): if not isinstance(config, dict):
err(f"config.yaml: root must be a mapping, got {type(config).__name__}") err(f"config.yaml: root must be a mapping, got {type(config).__name__}")
return return
for key in REQUIRED_KEYS:
if key not in config: # Schema-version dispatch. Validate the version field shape first
err(f"config.yaml: missing required key `{key}`") # so error messages are actionable.
runtime = config.get("runtime")
if runtime and runtime not in KNOWN_RUNTIMES:
warn(
f"config.yaml: runtime `{runtime}` not in known set "
f"{sorted(KNOWN_RUNTIMES)} — OK for custom runtimes; "
f"if canonical, add it to KNOWN_RUNTIMES in validate-workspace-template.py"
)
sv = config.get("template_schema_version") sv = config.get("template_schema_version")
if sv is not None and not isinstance(sv, int): if sv is None:
err("config.yaml: missing required key `template_schema_version`")
# Can't dispatch without a version. Don't fall through to v1
# checks — that would mask the missing-version error.
return
if not isinstance(sv, int):
err( err(
f"config.yaml: template_schema_version must be int, " f"config.yaml: template_schema_version must be int, "
f"got {type(sv).__name__}={sv!r}" f"got {type(sv).__name__}={sv!r}"
) )
return
unknown = set(config.keys()) - set(REQUIRED_KEYS) - set(OPTIONAL_KEYS) if sv in DEPRECATED_SCHEMA_VERSIONS:
if unknown: latest = max(KNOWN_SCHEMA_VERSIONS)
warn( warn(
f"config.yaml: unknown top-level keys {sorted(unknown)}" f"config.yaml: template_schema_version={sv} is deprecated; "
f"may be drift. If intentional, add them to OPTIONAL_KEYS." f"migrate to v{latest} via "
f"`python3 scripts/migrate-template.py --to {latest} .`. "
f"Support for v{sv} will be removed in a future cycle."
) )
elif sv not in KNOWN_SCHEMA_VERSIONS:
valid = sorted(KNOWN_SCHEMA_VERSIONS | DEPRECATED_SCHEMA_VERSIONS)
err(
f"config.yaml: template_schema_version={sv} is unknown — "
f"this validator understands {valid}. Either bump the "
f"validator (add a SCHEMA_V{sv} block) or correct the version."
)
return
SCHEMA_CHECKS[sv](config)
# ───────────────────────────────────────────────────────────── requirements.txt # ───────────────────────────────────────────────────────────── requirements.txt