schema.py
python
sha256:0e9549ec7b463911bc08b7d586dc320b1ac9b1f5c943ee7e3865dcc6cb0f6f83
chore(governance): sync handover+roadmap to 84db8c8 (drift:…
Human
1 day ago
| 1 | """Policy and manifest loaders (§K9.3 / §K9.4).""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | from pathlib import Path |
| 6 | from typing import Any |
| 7 | |
| 8 | import yaml |
| 9 | |
| 10 | from tools.checkpoints.ids import is_valid_step_id |
| 11 | from tools.checkpoints.types import ManifestState, PolicyState, StepDef, StepState |
| 12 | |
| 13 | |
| 14 | class CheckpointSchemaError(ValueError): |
| 15 | """Raised when policy or manifest shape is invalid.""" |
| 16 | |
| 17 | |
| 18 | def load_policy(path: Path) -> PolicyState: |
| 19 | """Load and validate ``policy/checkpoints.yaml``.""" |
| 20 | try: |
| 21 | raw = yaml.safe_load(path.read_text(encoding="utf-8")) |
| 22 | except OSError as exc: |
| 23 | raise CheckpointSchemaError(f"cannot read policy: {exc}") from exc |
| 24 | except yaml.YAMLError as exc: |
| 25 | raise CheckpointSchemaError(f"unparseable policy YAML: {exc}") from exc |
| 26 | if not isinstance(raw, dict): |
| 27 | raise CheckpointSchemaError("policy root must be a mapping") |
| 28 | |
| 29 | version = raw.get("version") |
| 30 | if not isinstance(version, int) or version != 1: |
| 31 | raise CheckpointSchemaError("policy version must be integer 1") |
| 32 | |
| 33 | tokens_raw = raw.get("placeholder_tokens", []) |
| 34 | if tokens_raw is None: |
| 35 | tokens_raw = [] |
| 36 | if not isinstance(tokens_raw, list) or not all(isinstance(x, str) for x in tokens_raw): |
| 37 | raise CheckpointSchemaError("placeholder_tokens must be a list of strings") |
| 38 | placeholder_tokens = list(tokens_raw) |
| 39 | |
| 40 | steps_raw = raw.get("steps") |
| 41 | if not isinstance(steps_raw, dict): |
| 42 | raise CheckpointSchemaError("policy steps must be a mapping") |
| 43 | steps: dict[str, StepDef] = {} |
| 44 | for step_id, step_raw in steps_raw.items(): |
| 45 | if not is_valid_step_id(step_id): |
| 46 | raise CheckpointSchemaError(f"invalid step_id {step_id!r}") |
| 47 | if not isinstance(step_raw, dict): |
| 48 | raise CheckpointSchemaError(f"step {step_id!r} must be a mapping") |
| 49 | verify_script = step_raw.get("verify_script") |
| 50 | if verify_script is not None and not isinstance(verify_script, str): |
| 51 | raise CheckpointSchemaError(f"step {step_id!r} verify_script must be a string") |
| 52 | description = step_raw.get("description") |
| 53 | if description is not None and not isinstance(description, str): |
| 54 | raise CheckpointSchemaError(f"step {step_id!r} description must be a string") |
| 55 | steps[step_id] = StepDef( |
| 56 | verify_script=verify_script or "", |
| 57 | description=description, |
| 58 | ) |
| 59 | |
| 60 | templates_raw = raw.get("templates") |
| 61 | if not isinstance(templates_raw, dict): |
| 62 | raise CheckpointSchemaError("policy templates must be a mapping") |
| 63 | templates: dict[str, list[str]] = {} |
| 64 | for template_id, template_raw in templates_raw.items(): |
| 65 | if not is_valid_step_id(template_id): |
| 66 | raise CheckpointSchemaError(f"invalid template_id {template_id!r}") |
| 67 | if not isinstance(template_raw, dict): |
| 68 | raise CheckpointSchemaError(f"template {template_id!r} must be a mapping") |
| 69 | steps_list = template_raw.get("steps") |
| 70 | if not isinstance(steps_list, list) or not steps_list: |
| 71 | raise CheckpointSchemaError(f"template {template_id!r} steps must be a non-empty list") |
| 72 | for sid in steps_list: |
| 73 | if not isinstance(sid, str) or not is_valid_step_id(sid): |
| 74 | raise CheckpointSchemaError(f"invalid step id in template {template_id!r}") |
| 75 | templates[template_id] = list(steps_list) |
| 76 | |
| 77 | overrides_raw = raw.get("overrides") or {} |
| 78 | if not isinstance(overrides_raw, dict): |
| 79 | raise CheckpointSchemaError("overrides must be a mapping") |
| 80 | overrides_default = overrides_raw.get("default") or {} |
| 81 | if not isinstance(overrides_default, dict): |
| 82 | raise CheckpointSchemaError("overrides.default must be a mapping") |
| 83 | overrides_by_template: dict[str, dict[str, Any]] = {} |
| 84 | for key, value in overrides_raw.items(): |
| 85 | if key == "default": |
| 86 | continue |
| 87 | if not isinstance(value, dict): |
| 88 | raise CheckpointSchemaError(f"overrides.{key} must be a mapping") |
| 89 | overrides_by_template[key] = dict(value) |
| 90 | |
| 91 | return PolicyState( |
| 92 | version=version, |
| 93 | placeholder_tokens=placeholder_tokens, |
| 94 | steps=steps, |
| 95 | templates=templates, |
| 96 | overrides_default=dict(overrides_default), |
| 97 | overrides_by_template=overrides_by_template, |
| 98 | source_path=str(path), |
| 99 | ) |
| 100 | |
| 101 | |
| 102 | def load_manifest(path: Path) -> ManifestState: |
| 103 | """Load and validate active manifest.""" |
| 104 | try: |
| 105 | raw = yaml.safe_load(path.read_text(encoding="utf-8")) |
| 106 | except OSError as exc: |
| 107 | raise CheckpointSchemaError(f"cannot read manifest: {exc}") from exc |
| 108 | except yaml.YAMLError as exc: |
| 109 | raise CheckpointSchemaError(f"unparseable manifest YAML: {exc}") from exc |
| 110 | if not isinstance(raw, dict): |
| 111 | raise CheckpointSchemaError("manifest root must be a mapping") |
| 112 | |
| 113 | schema_version = raw.get("schema_version") |
| 114 | if not isinstance(schema_version, int) or schema_version != 1: |
| 115 | raise CheckpointSchemaError("manifest schema_version must be integer 1") |
| 116 | |
| 117 | template_id = raw.get("template_id") |
| 118 | if not isinstance(template_id, str) or not is_valid_step_id(template_id): |
| 119 | raise CheckpointSchemaError("manifest template_id invalid") |
| 120 | |
| 121 | slug = raw.get("slug") |
| 122 | if not isinstance(slug, str) or not slug.strip(): |
| 123 | raise CheckpointSchemaError("manifest slug must be a non-empty string") |
| 124 | |
| 125 | current_step = raw.get("current_step") |
| 126 | if not isinstance(current_step, str) or not is_valid_step_id(current_step): |
| 127 | raise CheckpointSchemaError("manifest current_step invalid") |
| 128 | |
| 129 | meta = raw.get("meta") or {} |
| 130 | if not isinstance(meta, dict): |
| 131 | raise CheckpointSchemaError("manifest meta must be a mapping") |
| 132 | |
| 133 | steps_raw = raw.get("steps") |
| 134 | if not isinstance(steps_raw, dict): |
| 135 | raise CheckpointSchemaError("manifest steps must be a mapping") |
| 136 | steps: dict[str, StepState] = {} |
| 137 | for step_id, step_raw in steps_raw.items(): |
| 138 | if not isinstance(step_raw, dict): |
| 139 | raise CheckpointSchemaError(f"manifest step {step_id!r} must be a mapping") |
| 140 | verified = step_raw.get("verified", False) |
| 141 | if not isinstance(verified, bool): |
| 142 | raise CheckpointSchemaError(f"manifest step {step_id!r} verified must be boolean") |
| 143 | verified_at = step_raw.get("verified_at") |
| 144 | if verified_at is not None and not isinstance(verified_at, str): |
| 145 | raise CheckpointSchemaError(f"manifest step {step_id!r} verified_at must be string or null") |
| 146 | artifact_sha256 = step_raw.get("artifact_sha256") |
| 147 | if artifact_sha256 is not None and not isinstance(artifact_sha256, str): |
| 148 | raise CheckpointSchemaError( |
| 149 | f"manifest step {step_id!r} artifact_sha256 must be string or null" |
| 150 | ) |
| 151 | steps[step_id] = StepState( |
| 152 | verified=verified, |
| 153 | verified_at=verified_at, |
| 154 | artifact_sha256=artifact_sha256, |
| 155 | ) |
| 156 | |
| 157 | return ManifestState( |
| 158 | schema_version=schema_version, |
| 159 | template_id=template_id, |
| 160 | slug=slug, |
| 161 | current_step=current_step, |
| 162 | steps=steps, |
| 163 | meta=dict(meta), |
| 164 | source_path=str(path), |
| 165 | ) |
| 166 | |
| 167 | |
| 168 | def resolve_template_steps(policy: PolicyState, manifest: ManifestState) -> list[str]: |
| 169 | """Resolve ordered template list ``T``; raise on missing template or bad refs.""" |
| 170 | if manifest.template_id not in policy.templates: |
| 171 | raise CheckpointSchemaError(f"unknown template_id {manifest.template_id!r}") |
| 172 | template_steps = policy.templates[manifest.template_id] |
| 173 | for step_id in template_steps: |
| 174 | step_def = policy.steps.get(step_id) |
| 175 | if step_def is None or not step_def.verify_script.strip(): |
| 176 | raise CheckpointSchemaError( |
| 177 | f"template step {step_id!r} missing from policy.steps or empty verify_script" |
| 178 | ) |
| 179 | if manifest.current_step not in template_steps: |
| 180 | raise CheckpointSchemaError("manifest current_step not in template") |
| 181 | for step_id in template_steps: |
| 182 | if step_id not in manifest.steps: |
| 183 | raise CheckpointSchemaError(f"manifest missing step {step_id!r}") |
| 184 | return template_steps |
| 185 | |
| 186 | |
| 187 | def merge_overrides(policy: PolicyState, template_id: str) -> dict[str, Any]: |
| 188 | """Shallow merge overrides.default then template-specific.""" |
| 189 | merged = dict(policy.overrides_default) |
| 190 | if template_id in policy.overrides_by_template: |
| 191 | merged.update(policy.overrides_by_template[template_id]) |
| 192 | return merged |
| 193 | |
| 194 | |
| 195 | def manifest_to_yaml(manifest: ManifestState) -> str: |
| 196 | """Serialize manifest for atomic write.""" |
| 197 | data: dict[str, Any] = { |
| 198 | "schema_version": manifest.schema_version, |
| 199 | "template_id": manifest.template_id, |
| 200 | "slug": manifest.slug, |
| 201 | "current_step": manifest.current_step, |
| 202 | "meta": manifest.meta, |
| 203 | "steps": {}, |
| 204 | } |
| 205 | for step_id, state in manifest.steps.items(): |
| 206 | data["steps"][step_id] = { |
| 207 | "verified": state.verified, |
| 208 | "verified_at": state.verified_at, |
| 209 | "artifact_sha256": state.artifact_sha256, |
| 210 | } |
| 211 | return yaml.safe_dump(data, sort_keys=False, allow_unicode=True) |
| 212 | |
| 213 | |
| 214 | def render_progress(manifest: ManifestState, template_steps: list[str]) -> str: |
| 215 | """Deterministic PROGRESS.md renderer (§K9.4).""" |
| 216 | lines = [ |
| 217 | f"# Progress — {manifest.slug}", |
| 218 | "", |
| 219 | f"Template: `{manifest.template_id}`", |
| 220 | f"Current step: `{manifest.current_step}`", |
| 221 | "", |
| 222 | "## Steps", |
| 223 | "", |
| 224 | ] |
| 225 | for step_id in template_steps: |
| 226 | state = manifest.steps.get(step_id) |
| 227 | if state is None: |
| 228 | mark = "?" |
| 229 | elif state.verified: |
| 230 | mark = "✓" |
| 231 | else: |
| 232 | mark = "·" |
| 233 | lines.append(f"- [{mark}] `{step_id}`") |
| 234 | lines.append("") |
| 235 | return "\n".join(lines) |
File History
1 commit
sha256:6abcf1fa82a7a621ccbc945f19acdba5bc0db54569599404a1452fb4a096a199
fix(ISR): default require_independent_second_reviewer to require
Human
minor
⚠
1 day ago