127 lines
4.5 KiB
Python
Executable File
127 lines
4.5 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Render reviewable Flux artifacts for a published Hermes agent image."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import difflib
|
|
import json
|
|
import re
|
|
from pathlib import Path
|
|
|
|
|
|
DEFAULT_IMAGE = "registry.bstein.dev/bstein/hermes-agent"
|
|
DIGEST_PATTERN = re.compile(r"^sha256:[0-9a-f]{64}$")
|
|
REVISION_PATTERN = re.compile(r"^[0-9a-f]{40}$")
|
|
|
|
|
|
def _validated(value: str, pattern: re.Pattern[str], label: str) -> str:
|
|
"""Return a normalized value when it matches the release contract."""
|
|
normalized = value.strip()
|
|
if not pattern.fullmatch(normalized):
|
|
raise ValueError(f"invalid {label}: expected {pattern.pattern}")
|
|
return normalized
|
|
|
|
|
|
def render_kustomization(source: str, digest: str, image: str = DEFAULT_IMAGE) -> str:
|
|
"""Replace exactly one matching Kustomize image digest without reformatting."""
|
|
digest = _validated(digest, DIGEST_PATTERN, "image digest")
|
|
lines = source.splitlines(keepends=True)
|
|
matches: list[int] = []
|
|
|
|
for index, line in enumerate(lines):
|
|
if line.strip() != f"- name: {image}":
|
|
continue
|
|
name_indent = len(line) - len(line.lstrip())
|
|
for candidate_index in range(index + 1, len(lines)):
|
|
candidate = lines[candidate_index]
|
|
stripped = candidate.strip()
|
|
candidate_indent = len(candidate) - len(candidate.lstrip())
|
|
if stripped.startswith("- name:") and candidate_indent == name_indent:
|
|
break
|
|
if stripped.startswith("digest:") and candidate_indent > name_indent:
|
|
matches.append(candidate_index)
|
|
break
|
|
|
|
if len(matches) != 1:
|
|
raise ValueError(
|
|
f"expected exactly one digest for image {image!r}; found {len(matches)}"
|
|
)
|
|
|
|
index = matches[0]
|
|
newline = "\n" if lines[index].endswith("\n") else ""
|
|
prefix = lines[index][: len(lines[index]) - len(lines[index].lstrip())]
|
|
lines[index] = f"{prefix}digest: {digest}{newline}"
|
|
return "".join(lines)
|
|
|
|
|
|
def write_release_artifacts(
|
|
*,
|
|
digest: str,
|
|
source_revision: str,
|
|
destination: str,
|
|
kustomization: Path,
|
|
output_dir: Path,
|
|
) -> dict[str, str]:
|
|
"""Write a rendered manifest, patch, and credential-free release metadata."""
|
|
digest = _validated(digest, DIGEST_PATTERN, "image digest")
|
|
source_revision = _validated(
|
|
source_revision, REVISION_PATTERN, "source revision"
|
|
)
|
|
if destination != f"{DEFAULT_IMAGE}:git-{source_revision}":
|
|
raise ValueError("destination must be the immutable git-<revision> tag")
|
|
|
|
source = kustomization.read_text(encoding="utf-8")
|
|
rendered = render_kustomization(source, digest)
|
|
relative_name = kustomization.name
|
|
patch = "".join(
|
|
difflib.unified_diff(
|
|
source.splitlines(keepends=True),
|
|
rendered.splitlines(keepends=True),
|
|
fromfile=f"a/services/hermes/{relative_name}",
|
|
tofile=f"b/services/hermes/{relative_name}",
|
|
)
|
|
)
|
|
if not patch:
|
|
raise ValueError("published digest already matches the Flux manifest")
|
|
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
(output_dir / "hermes-kustomization.yaml").write_text(
|
|
rendered, encoding="utf-8"
|
|
)
|
|
(output_dir / "hermes-image-update.patch").write_text(patch, encoding="utf-8")
|
|
metadata = {
|
|
"digest": digest,
|
|
"flux_image": f"{DEFAULT_IMAGE}@{digest}",
|
|
"image": DEFAULT_IMAGE,
|
|
"published_tag": destination,
|
|
"source_revision": source_revision,
|
|
}
|
|
(output_dir / "hermes-agent-image.json").write_text(
|
|
json.dumps(metadata, indent=2, sort_keys=True) + "\n", encoding="utf-8"
|
|
)
|
|
return metadata
|
|
|
|
|
|
def main() -> int:
|
|
"""Validate Kaniko output and render artifacts for the reviewed Flux PR."""
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--digest-file", required=True, type=Path)
|
|
parser.add_argument("--source-revision", required=True)
|
|
parser.add_argument("--destination", required=True)
|
|
parser.add_argument("--kustomization", required=True, type=Path)
|
|
parser.add_argument("--output-dir", required=True, type=Path)
|
|
args = parser.parse_args()
|
|
write_release_artifacts(
|
|
digest=args.digest_file.read_text(encoding="utf-8"),
|
|
source_revision=args.source_revision,
|
|
destination=args.destination,
|
|
kustomization=args.kustomization,
|
|
output_dir=args.output_dir,
|
|
)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__": # pragma: no cover - exercised through main()
|
|
raise SystemExit(main())
|