feat(samever): step back to older base when last-green canonical == head version

resolve_upgrade_base now reads the head's published version (abra.head_compose_version,
the coop-cloud.<stack>.version label) and, when the last-green warm-canonical version
equals it, steps back to the newest published version strictly older than head instead
of deploying a same-version no-op. warm_reconcile gains version_key + newest_older_version
(single coop-cloud ordering source; sort_versions refactored onto version_key, no behavior
change). Skip only when no older published predecessor exists. Step-back returns kind=version
so it inherits F1d-2 pinned-tag checkout. Extends tests/unit/test_upgrade_base.py (13 pass).
This commit is contained in:
autonomic-bot
2026-06-17 04:24:14 +00:00
parent 279d84d229
commit b29bb3f804
4 changed files with 175 additions and 9 deletions

View File

@ -11,6 +11,7 @@ from __future__ import annotations
import json
import os
import re
import subprocess
ABRA = "abra"
@ -34,6 +35,29 @@ def recipe_dir(recipe: str) -> str:
return os.path.join(abra_dir(), "recipes", recipe)
_VERSION_LABEL_RE = re.compile(r"coop-cloud\.[^.\s]*\.version=([^\s\"']+)")
def head_compose_version(recipe: str) -> str | None:
"""The published version of the recipe's on-disk compose.yml (the head checkout under test):
the value of the `coop-cloud.<stack>.version` label, e.g. "1.0.0+3.5.3". None if the file is
unreadable or carries no version label.
Used by the upgrade-base resolver (phase samever) to compare the head's declared version against
the last-green warm-canonical version: when they are equal, deploying the canonical as the base
would be a vacuous same-version no-op, so the resolver steps back to an older published version.
`${STACK_NAME}` is a literal in the file (abra interpolates it at deploy time, not on disk), so
the regex matches it as the stack segment without needing interpolation."""
path = os.path.join(recipe_dir(recipe), "compose.yml")
try:
with open(path) as f:
text = f.read()
except OSError:
return None
m = _VERSION_LABEL_RE.search(text)
return m.group(1).strip() if m else None
def _run_pty(
args: list[str], timeout: int = 900, check: bool = True
) -> subprocess.CompletedProcess: