# Copyright Kevin Deldycke <[email protected]> and contributors.
#
# This program is Free Software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
"""Registry of the cooldown-respecting dependency updaters, and their driver.
The five `sync-*` dependency bumpers (`sync-dep-sources`, `sync-uv-lock`,
`sync-tool-versions`, `sync-action-pins`, `sync-workflow-pins`) share a shape:
discover the latest eligible upstream version, gated by the
`[tool.repomatic] minimum-release-age` cooldown (or uv's `exclude-newer` for
the lock), then rewrite the pin. This module turns that shape into data: one
{class}`SyncOperation` per bumper, in {data}`SYNC_OPERATIONS`. The thin
`sync-*` commands and the aggregate `sync-deps` command in
{mod}`repomatic.cli.deps` all drive the operations through it.
```{rubric} Resolve then apply
```
Each operation splits into a read phase and a write phase so `sync-deps` can run
the slow, network-bound discovery for every operation concurrently, then write
serially:
- {attr}`SyncOperation.resolve` performs the network discovery and computes the
new file contents in memory, returning a {class}`SyncPlan`. It does not touch
the repository, so the resolves are safe to run in parallel.
- {attr}`SyncOperation.apply` writes the planned contents. Two of the five
operations rewrite `.github/workflows/*.yaml` (action pins and workflow
literals both live there), so applies must run serially.
`sync-uv-lock` and `sync-dep-sources` are the documented exceptions: their
discovery *is* a mutation (`uv lock` rewrites `uv.lock`), so their
{attr}`SyncOperation.resolve` writes during the parallel phase and their
{attr}`SyncOperation.apply` is a no-op. Their shared write domain (`uv.lock`,
`pyproject.toml`) is disjoint from every other operation's, and the two are
serialized against each other through `_UV_PROJECT_MUTEX`. A `--dry-run`
resolve snapshots and restores the mutated files so the preview leaves no
trace.
The datasource adapters, version selection, and pure string rewriters live in
{mod}`repomatic.deps.pins` and {mod}`repomatic.deps.cooldown`; this module
composes them with the file I/O and checksum recompute. The markdown report of
a plan is rendered here too ({func}`render_plan_markdown`), since the CI jobs
read it. The terminal output stays in the CLI.
"""
from __future__ import annotations
import json
import logging
import subprocess
import threading
from contextlib import contextmanager
from dataclasses import dataclass, field
from datetime import UTC, date, datetime
from functools import cached_property, partial
from pathlib import Path
from click_extra import OperationTrail, run_jobs
from ..components import BUNDLED_VERBATIM_TARGETS
from ..file_ops import write_if_changed
from ..github.releases import (
ReleaseNotesRequest,
fetch_github_release_notes,
resolve_tag_to_sha,
)
from ..github.workflow_files import github_yaml_paths
from ..http import FetchError, get_text
from ..humanize import format_countdown, utc_midnight
from ..init.tool_config import init_config
from ..init.upgrade import upgrade_invite
from ..pypi import PYPI_PACKAGE_URL, get_source_url
from ..templating import PrTemplate
from ..tooling import registry as tool_registry
from ..tooling.checksums import update_registry_checksums
from ..tooling.registry import TOOL_REGISTRY
from ..upstream import (
DEFAULT_REPO,
UPSTREAM_PACKAGE,
UPSTREAM_REPO_SLUGS,
is_source_repo,
)
from ..versions import is_newer
from .cooldown import (
UvSettings,
active_forecasts,
compute_held_back_packages,
sync_uv_lock,
)
from .lockfile import LockFile, diff_lock_versions
from .pins import (
MIN_AGE_HELD_BACK_NOTE,
SETUP_UV_PACKAGE,
SETUP_UV_SLUG,
Ecosystem,
apply_action_pins,
apply_workflow_literals,
find_action_pins,
find_upstream_ref_versions,
find_workflow_literals,
format_cooldown_note,
github_candidates,
parse_min_age,
pin_inside_cooldown,
select_held_back,
select_latest,
set_tool_version,
set_with_package_version,
setup_uv_verified_versions,
tool_candidates,
)
from .report import (
EXCLUDE_NEWER_HELD_BACK_NOTE,
BypassForecast,
HeldBackPackage,
build_comparison_urls,
build_held_back,
fetch_release_notes,
format_bypass_section,
format_diff_table,
format_exclude_newer_note,
format_held_back_table,
format_release_notes,
pypi_name_urls,
)
from .requirements import read_project
from .sources import (
ReleaseSwap,
apply_release_swaps,
find_ready_swaps,
format_swap_section,
tracked_git_overrides,
)
from .uv import run_uv
TYPE_CHECKING = False
if TYPE_CHECKING:
from collections.abc import Callable, Generator, Iterable, Mapping, Sequence
from datetime import timedelta
from typing import Any
from ..config import Config
from .pins import Candidate
ACTIONLINT_MATCHER_URL = (
"https://raw.githubusercontent.com/rhysd/actionlint/refs/tags/v{version}"
"/.github/actionlint-matcher.json"
)
"""Where actionlint keeps its problem matcher, at the tag of a release."""
DEPENDENCY_LABEL = "🔗 dependencies"
"""GitHub label applied to every dependency-update PR.
Shared by all five bumpers so a single label filters the whole family. Workflow
YAML cannot import Python, so `autofix.yaml` repeats this string literally, and
the labeller's own rule tables ({data}`repomatic.github.labels.DEFAULT_CONTENT_RULES`
and {data}`~repomatic.github.labels.DEFAULT_FILE_RULES`) key their dependency rules on
the same spelling. `tests/test_sync_ops.py` asserts both copies match this
constant, and `tests/test_labels.py` that it names a label `labels.toml`
actually defines: applying an unknown label fails the `gh` call outright, so a
rename in the registry has to reach all of them at once.
"""
_UV_PROJECT_MUTEX = threading.Lock()
"""Serializes the two resolves that mutate the uv project files.
`sync-uv-lock` and `sync-dep-sources` both rewrite `pyproject.toml` and run
`uv lock` during their resolve phase, while `sync-deps` fans the resolves out
concurrently. Interleaving two lock runs (or a snapshot/restore pair) on the
same project corrupts both, so the two resolves take this mutex for their
whole duration.
"""
def _pinnable_files() -> dict[Path, str]:
"""Read the `.github/` workflow and action files whose pins are bumpable.
Downstream, drops the files `repomatic init` deploys verbatim
({data}`~repomatic.components.BUNDLED_VERBATIM_TARGETS`): their `uses:` pins are
owned by the bundled template, so a bump here is undone by the next
`sync-repomatic` and the two pull requests ping-pong. This is the per-file
counterpart to the per-slug {data}`~repomatic.upstream.UPSTREAM_REPO_SLUGS`
skip in `_resolve_action_pins`.
In the source repo the exclusion lifts: there the bundled template is a symlink
to the in-tree file, so init rewrites nothing and the pin is a normal
source-of-truth ref that upstream keeps bumping (see
{func}`~repomatic.upstream.is_source_repo`).
"""
in_source_repo = is_source_repo(Path.cwd())
return {
path: path.read_text(encoding="UTF-8")
for path in github_yaml_paths(Path())
if in_source_repo or path.as_posix() not in BUNDLED_VERBATIM_TARGETS
}
def _sync_actionlint_matcher(version: str) -> None:
"""Refresh the bundled actionlint problem matcher from the tag of *version*.
The matcher reads the output format of actionlint, so the copy `repomatic
run actionlint` registers moves with the pinned release, in the pull
request of the bump. A fetch that fails keeps the previous copy, with a
warning: the format has not changed across releases, and the bump of the
tool itself is still worth proposing.
The copy is indented with tabs, the way the `format-json` job writes it.
"""
spec = TOOL_REGISTRY["actionlint"]
assert spec.problem_matcher
url = ACTIONLINT_MATCHER_URL.format(version=version)
try:
matcher = json.loads(get_text(url))
except (FetchError, json.JSONDecodeError) as error:
logging.warning(
f"Cannot fetch the actionlint {version} problem matcher, keeping the"
f" previous copy: {error}"
)
return
path = Path(tool_registry.__file__).parents[1] / "data" / spec.problem_matcher
write_if_changed(path, json.dumps(matcher, indent="\t") + "\n")
@contextmanager
def _snapshot(*paths: Path, restore: bool) -> Generator[Callable[[], None], None, None]:
"""Put *paths* back to their bytes on entry, when the block fails.
The resolves of the uv-project pair mutate the files they plan from, so a
failure mid-resolve must not leave a half-mutated project (synced pins
with a stale lock) for the CI job to commit.
:param paths: The files to snapshot. A file that does not exist is left
alone.
:param restore: Also put the files back when the block succeeds: a
`--dry-run` resolve leaves no trace.
:return: A callable restoring the snapshot, for a block that gives up on
its own without raising.
"""
saved = {path: path.read_bytes() for path in paths if path.exists()}
def put_back() -> None:
for path, content in saved.items():
path.write_bytes(content)
try:
yield put_back
except BaseException:
put_back()
raise
if restore:
put_back()
[docs]
@dataclass
class ResolveContext:
"""Inputs shared by every {attr}`SyncOperation.resolve`.
Each operation reads the subset it needs. The cooldown is derived from
*config* (`minimum-release-age` for the version-sync trio, `exclude-newer`
from the lock for `sync-uv-lock`).
"""
config: Config
"""The resolved `[tool.repomatic]` configuration."""
today: date
"""Reference date for the cooldown computation, fixed once per run."""
release_notes: bool = False
"""Fetch upstream release notes and append them to the report."""
held_back: bool = True
"""Report newer releases withheld only by the cooldown."""
dry_run: bool = False
"""Plan without persisting: restore any files the resolve had to mutate."""
lockfile: Path = field(default_factory=lambda: Path("uv.lock"))
"""Path to `uv.lock` for `sync-uv-lock`."""
jobs: int = 1
"""Worker threads for the per-item datasource fetches.
The CLI resolves it from `--jobs` on its own thread, since a worker thread
running a resolve has no Click context to read it from.
"""
[docs]
@cached_property
def min_age(self) -> timedelta:
"""The `minimum-release-age` cooldown, parsed."""
return parse_min_age(self.config.minimum_release_age)
[docs]
@cached_property
def pinnable_files(self) -> dict[Path, str]:
"""The `.github/` files the two pin updaters scan, read once per run."""
return _pinnable_files()
[docs]
@dataclass
class SyncPlan:
"""The resolved, not-yet-written outcome of one operation's read phase.
Carries everything {attr}`SyncOperation.apply` needs to write the changes
and everything the CLI needs to render the terminal table and the markdown
PR body, so the write and the rendering never re-resolve.
"""
operation: str
"""The operation name (`sync-uv-lock`, …)."""
subject: str
"""Header for the first table column (`Package`, `Tool`, `Action`)."""
heading: str
"""Noun after `## 🆙 ` in the diff table (`Updated tools`, …)."""
changes: list[tuple[str, str, str]] = field(default_factory=list)
"""Applied `(name, old, new)` triples, in the order the report renders."""
dates: dict[str, str] = field(default_factory=dict)
"""Name to release/upload date (`YYYY-MM-DD` or ISO 8601) for the table."""
released_overrides: dict[str, str] = field(default_factory=dict)
"""Name to literal markdown replacing its "Released" table cell.
Marks rows whose version was decided outside the cooldown-checked release
listing (the upstream toolkit's lockstep-aligned pin), so the table shows
the exemption instead of a blank cell.
"""
change_labels: dict[str, str] = field(default_factory=dict)
"""Name to literal markdown labelling its "Change" table cell.
Marks the uv pin stepped back onto the checksum table of the pinned
`setup-uv` (see `_gate_uv_on_checksums`), which the arrow alone renders
like any bump.
"""
name_urls: dict[str, str] = field(default_factory=dict)
"""Name to the URL its table cell links to (PyPI, GitHub, npm)."""
comparison_urls: dict[str, str] = field(default_factory=dict)
"""Name to a compare URL linked on the change cell."""
held_back: list[HeldBackPackage] = field(default_factory=list)
"""Newer releases withheld only by the cooldown."""
held_back_name_urls: dict[str, str] = field(default_factory=dict)
"""Name to URL for the held-back section."""
held_back_note: str = MIN_AGE_HELD_BACK_NOTE
"""Intro paragraph for the held-back section (cooldown wording)."""
notes_section: str = ""
"""Pre-rendered release-notes markdown, or empty."""
upgrade_section: str = ""
"""Pre-rendered invitation to review an upstream toolkit bump, or empty."""
cooldown_note: str = ""
"""Pre-rendered cooldown-cutoff sentence shown above the diff table."""
cutoff: date | None = None
"""Effective `minimum-release-age` cutoff, for the terminal report."""
reference_date: date | None = None
"""Reference date for the table's relative "Released" hints (the run date)."""
file_writes: dict[Path, str] = field(default_factory=dict)
"""Path to its new full text, as computed at resolve time.
`sync-tool-versions` writes this text as is. The two `.github/` pin
updaters only read which files it names: their apply replays the rewrite on
current disk state through {attr}`rebase`.
"""
self_pin_exemptions: list[str] = field(default_factory=list)
"""Workflow files that gained a missing self-pin cooldown exemption.
Names only the files whose *sole* edit was the splice, since a file that
also moved a version is already reported through {attr}`changes`. Kept as a
separate list for the same reason {attr}`UvProjectExtras.frozen_bypasses`
is: the rewrite has no `(name, old, new)` triple to render, yet it still
produced a hunk the report has to explain. Without it a splice-only run
reads as "nothing to update" and the write is dropped, which is what let an
exemption-less downstream pin sit broken indefinitely: the backfill only
ever landed on a run that happened to move the version too.
"""
rebase: Callable[[str], tuple[str, list[tuple[str, str, str]]]] | None = None
"""Replay this operation's rewriter against a file's current text.
Set by `_plan_file_rewrites`. Applies run serially after every
resolve finished, and the two `.github/` pin updaters routinely plan
rewrites of the same workflow files from the same pre-apply snapshot:
writing {attr}`file_writes` verbatim would silently revert whichever
sibling applied first. The closure re-runs the pure rewriter on whatever
is on disk at apply time instead.
"""
tool_versions: ToolVersionExtras = field(default_factory=ToolVersionExtras)
"""`sync-tool-versions` write extras (checksum recompute, problem matcher)."""
uv_project: UvProjectExtras = field(default_factory=UvProjectExtras)
"""`sync-uv-lock` and `sync-dep-sources` rendering extras."""
@property
def has_changes(self) -> bool:
"""Whether the operation found anything to update.
Cooldown-bypass edits count: a run that only prunes or freezes
`exclude-newer-package` entries still rewrites `pyproject.toml` and
must produce a report explaining that hunk. A run that only splices a
missing self-pin exemption into a workflow counts for the same reason.
"""
return bool(
self.changes
or self.self_pin_exemptions
or self.uv_project.pruned_bypasses
or self.uv_project.frozen_bypasses
)
[docs]
def note_cooldown(self, age_label: str, min_age: timedelta, today: date) -> None:
"""Record the cooldown cutoff and its rendered diff-table note.
No-op fields (a `None` cutoff, an empty note) when the cooldown is
disabled (`0 days` or unparsable).
"""
self.cutoff = today - min_age if min_age else None
self.cooldown_note = (
format_cooldown_note(age_label, self.cutoff)
if self.cutoff is not None
else ""
)
def _track_held_back(
plan: SyncPlan,
rc: ResolveContext,
name: str,
url: str,
candidates: list[Candidate],
latest: Candidate | None,
pinned: str,
) -> None:
"""Record on *plan* the release withheld from *name* only by the cooldown.
Held-back covers every scanned item, even ones not bumped this run, so the
section shows the full cooldown pipeline. The probe compares against the
version this run settles on: *latest* when it beats *pinned*, else
*pinned*. Skipped when the caller opted out (`rc.held_back` off).
"""
if not rc.held_back:
return
final_version = (
latest.version
if latest is not None and is_newer(latest.version, pinned)
else pinned
)
withheld = select_held_back(candidates, final_version, rc.min_age, rc.today)
if withheld is not None:
plan.held_back.append(
build_held_back(
name,
final_version,
withheld.version,
withheld.date,
rc.min_age,
rc.today,
)
)
plan.held_back_name_urls[name] = url
def _finish_uv_plan(
plan: SyncPlan,
rc: ResolveContext,
lock: LockFile,
settings: UvSettings,
) -> None:
"""Attach the held-back probe, release notes and compare URLs to a uv plan.
The shared tail of the uv-project pair. Both operations end by asking uv the
same two follow-up questions about the lock they just wrote, so the wiring
(and the reason each step is gated) lives here once.
:param plan: The plan being resolved, mutated in place.
:param rc: The resolve inputs, supplying the opt-in flags and the lock path.
:param lock: The lock the operation just wrote.
:param settings: The project's `[tool.uv]` settings, after the operation.
"""
# The probe runs a second uv resolution, so skip it with nothing to report:
# the caller already gates `held_back` on a consumer being present (terminal
# table or file output). Bypass edits count as changes: a prune may leave
# newer releases cooldown-held, and this is the report that explains them.
if rc.held_back and plan.has_changes:
plan.held_back = compute_held_back_packages(
rc.lockfile, lock, settings.entries()
)
plan.held_back_name_urls = pypi_name_urls(pkg.name for pkg in plan.held_back)
notes: dict[str, tuple[str, list[tuple[str, str]]]] = {}
if rc.release_notes and plan.changes:
notes = fetch_release_notes(plan.changes)
plan.notes_section = format_release_notes(notes)
plan.comparison_urls = build_comparison_urls(plan.changes, notes)
def _resolve_uv_lock(rc: ResolveContext) -> SyncPlan:
"""Re-lock dependencies and roll cooldown overrides forward.
Unlike the version-sync trio, the discovery here is the mutation:
`uv lock --upgrade` rewrites `uv.lock` in place. The new contents are left
on disk (the write domain is shared only with `sync-dep-sources`, guarded
by {data}`_UV_PROJECT_MUTEX`), so {func}`_apply_uv_lock` is a no-op. A
`--dry-run` resolve snapshots `uv.lock` and `pyproject.toml` up front and
restores them once done.
"""
plan = SyncPlan(
operation="sync-uv-lock",
subject="Package",
heading="Updated packages",
held_back_note=EXCLUDE_NEWER_HELD_BACK_NOTE,
reference_date=rc.today,
)
lockfile = rc.lockfile
pyproject_path = lockfile.parent / "pyproject.toml"
with _UV_PROJECT_MUTEX, _snapshot(lockfile, pyproject_path, restore=rc.dry_run):
# Sync repomatic's canonical [tool.uv] policy pins from the bundled
# template before re-locking, so a bumped exclude-newer feeds the
# resolution and a bumped required-version rides in the same change.
merged = init_config("uv", pyproject_path) if pyproject_path.exists() else None
plan.uv_project.pins_synced = merged is not None
settings = UvSettings.load(pyproject_path, text=merged)
result = sync_uv_lock(lockfile, settings)
plan.changes = result.changes
plan.dates = result.lock.upload_times
plan.uv_project.exclude_newer = result.lock.exclude_newer
plan.cooldown_note = format_exclude_newer_note(result.lock.exclude_newer)
plan.name_urls = pypi_name_urls(name for name, _old, _new in result.changes)
plan.uv_project.pruned_bypasses = result.pruned_bypasses
plan.uv_project.frozen_bypasses = result.frozen_bypasses
plan.uv_project.stale_bypass_comments = result.stale_bypass_comments
plan.uv_project.bypass_forecasts = result.bypass_forecasts
_finish_uv_plan(plan, rc, result.lock, settings)
return plan
def _apply_uv_lock(plan: SyncPlan) -> None:
"""No-op: {func}`_resolve_uv_lock` already wrote `uv.lock`."""
def _resolve_dep_sources(rc: ResolveContext) -> SyncPlan:
"""Swap git-tracked dependencies whose awaited release shipped.
Like `sync-uv-lock`, the discovery is the mutation: the `pyproject.toml`
rewrite and the `uv lock` run happen here, guarded by
{data}`_UV_PROJECT_MUTEX`, and {func}`_apply_dep_sources` is a no-op.
Every failure path restores the snapshots taken up front: a swap is
all-or-nothing, so a resolution conflict or a lock landing on the wrong
version degrades to "nothing to report" instead of a half-applied swap.
"""
plan = SyncPlan(
operation="sync-dep-sources",
subject="Package",
heading="Updated packages",
held_back_note=EXCLUDE_NEWER_HELD_BACK_NOTE,
reference_date=rc.today,
)
lockfile = rc.lockfile
pyproject_path = lockfile.parent / "pyproject.toml"
if not pyproject_path.exists():
return plan
# The probe also runs under the mutex: the sibling `sync-uv-lock` resolve
# rewrites `pyproject.toml` (pin sync, prune, freeze) and reading a
# half-written file would corrupt the swap decision.
with _UV_PROJECT_MUTEX:
settings = UvSettings.load(pyproject_path)
swaps = find_ready_swaps(settings.doc)
if not swaps:
return plan
before = LockFile.load(lockfile)
with _snapshot(lockfile, pyproject_path, restore=rc.dry_run) as put_back:
apply_release_swaps(settings.doc, swaps)
settings.upsert({swap.name: swap.freeze_cutoff for swap in swaps})
settings.save()
try:
run_uv(settings.lock_command(), cwd=lockfile.parent)
except subprocess.CalledProcessError as error:
# A conflicting constraint elsewhere in the tree: not this
# run's call to untangle. Leave the project as found and report
# nothing.
put_back()
logging.warning("Dependency source swap failed to resolve; skipped.")
logging.debug(str(error))
return plan
post = LockFile.load(lockfile)
missed = [
swap.name
for swap in swaps
if post.versions.get(swap.name) != swap.release
]
if missed:
put_back()
logging.warning(
f"Swap landed on unexpected versions for {', '.join(missed)};"
" restored the project untouched."
)
return plan
plan.uv_project.source_swaps = swaps
plan.changes = diff_lock_versions(before.versions, post.versions)
plan.dates = post.upload_times
plan.uv_project.exclude_newer = post.exclude_newer
plan.cooldown_note = format_exclude_newer_note(post.exclude_newer)
plan.name_urls = pypi_name_urls(name for name, _old, _new in plan.changes)
# The fresh freezes are this run's news: they render as 📌 rows.
plan.uv_project.frozen_bypasses = [swap.name for swap in swaps]
plan.uv_project.bypass_forecasts = active_forecasts(
settings.entries(), post, datetime.now(UTC)
)
_finish_uv_plan(plan, rc, post, settings)
return plan
def _apply_dep_sources(plan: SyncPlan) -> None:
"""No-op: {func}`_resolve_dep_sources` already wrote the swap."""
def _pinned_with_packages() -> list[tuple[str, str]]:
"""Every `package==version` pin declared in a tool's `with_packages`.
These ride along in a tool's `uvx` environment (plugin sets, and the engine a
plugin shells out to) and never surface as a `repomatic run` tool of their
own, so nothing else in the bump path looks at them.
When two tools pin the same package the *lowest* version wins the lookup, so
a bump is proposed as long as any one of them trails the latest release;
{func}`~repomatic.deps.pins.set_with_package_version` then rewrites every
occurrence, converging them.
:return: Sorted `(package, pinned_version)` pairs.
"""
pins: dict[str, str] = {}
for spec in TOOL_REGISTRY.values():
for package, version in spec.with_package_pins:
if package not in pins or is_newer(pins[package], version):
pins[package] = version
return sorted(pins.items())
def _resolve_tool_versions(rc: ResolveContext) -> SyncPlan:
"""Bump every `repomatic run` tool to its latest eligible release."""
plan = SyncPlan(
operation="sync-tool-versions",
subject="Tool",
heading="Updated tools",
reference_date=rc.today,
)
registry_path = Path(tool_registry.__file__)
content = registry_path.read_text(encoding="UTF-8")
# Every datasource read is independent, and they are the slow part: fan
# them out, then sweep the answers in order.
tools = sorted(TOOL_REGISTRY.items())
packages = _pinned_with_packages()
tool_releases = list(
run_jobs(tool_candidates, [spec for _name, spec in tools], jobs=rc.jobs)
)
package_releases = list(
run_jobs(
Ecosystem.PYPI.candidates,
[package for package, _pinned in packages],
jobs=rc.jobs,
)
)
changes: list[tuple[str, str, str]] = []
overrides: dict[str, str] = {}
for (name, spec), candidates in zip(tools, tool_releases):
latest = select_latest(candidates, rc.min_age, rc.today)
_track_held_back(
plan, rc, name, spec.datasource_url, candidates, latest, spec.version
)
if latest is None or not is_newer(latest.version, spec.version):
continue
content = set_tool_version(content, name, latest.version)
changes.append((name, spec.version, latest.version))
plan.dates[name] = latest.date
overrides[name] = latest.version
# Second pass over the packages pinned *alongside* a tool. Without it they
# drift silently: a plugin pin ages out, or a second copy of a tool the main
# registry has already moved on from goes stale.
package_urls: dict[str, str] = {}
package_notes: list[ReleaseNotesRequest] = []
for (package, pinned), candidates in zip(packages, package_releases):
pypi_url = PYPI_PACKAGE_URL.format(package=package)
latest = select_latest(candidates, rc.min_age, rc.today)
_track_held_back(plan, rc, package, pypi_url, candidates, latest, pinned)
if latest is None or not is_newer(latest.version, pinned):
continue
content = set_with_package_version(content, package, latest.version)
changes.append((package, pinned, latest.version))
plan.dates[package] = latest.date
# Prefer the GitHub repository, matching ToolSpec.datasource_url's order,
# and fall back to the PyPI project page for a package declaring none.
# A resolved repo also earns the package release notes: no `tag_pattern`
# to consult here, so the default `vX.Y.Z` scheme applies (it also passes
# a bare `X.Y.Z` tag through untouched).
source = get_source_url(package)
package_urls[package] = source or pypi_url
if source:
package_notes.append(
ReleaseNotesRequest(package, source, pinned, latest.version)
)
plan.changes = changes
plan.note_cooldown(rc.config.minimum_release_age, rc.min_age, rc.today)
if not changes:
return plan
plan.file_writes = {registry_path: content}
plan.tool_versions.binary_overrides = {
name: version
for name, version in overrides.items()
if TOOL_REGISTRY[name].binary is not None
}
plan.tool_versions.actionlint_version = overrides.get("actionlint")
plan.tool_versions.jobs = rc.jobs
# `changes` now mixes registry tools with bare `with_packages` pins, so every
# TOOL_REGISTRY lookup below has to tolerate a name that is not a tool.
plan.name_urls = {
name: TOOL_REGISTRY[name].datasource_url
for name, _old, _new in changes
if name in TOOL_REGISTRY
and (TOOL_REGISTRY[name].source_url or TOOL_REGISTRY[name].npm is not None)
} | package_urls
if rc.release_notes:
tool_notes = [
ReleaseNotesRequest(name, src, old, new, TOOL_REGISTRY[name].tag_pattern)
for name, old, new in changes
if name in TOOL_REGISTRY
and (src := TOOL_REGISTRY[name].source_url)
and "github.com" in src
]
plan.notes_section = format_release_notes(
fetch_github_release_notes(tool_notes + package_notes)
)
return plan
def _apply_tool_versions(plan: SyncPlan) -> None:
"""Rewrite `tooling/registry.py`, recompute checksums, refresh the matcher."""
if not plan.has_changes:
return
for path, content in plan.file_writes.items():
path.write_text(content, encoding="UTF-8")
# Recompute checksums and VERSIONS stamps for bumped binary tools in
# the same pass, downloading at the new versions: the in-memory
# registry still holds the pre-bump versions, since the source was
# edited, not reimported.
if plan.tool_versions.binary_overrides:
update_registry_checksums(
path,
version_overrides=plan.tool_versions.binary_overrides,
jobs=plan.tool_versions.jobs,
)
if plan.tool_versions.actionlint_version:
_sync_actionlint_matcher(plan.tool_versions.actionlint_version)
def _widest_changes(
changes: Iterable[tuple[str, str, str]],
) -> list[tuple[str, str, str]]:
"""Collapse a name updated from several versions onto its widest range.
Files can pin the same action or package at different versions, but one
run converges them all onto a single resolved target, so a name's
`(name, old, new)` triples differ only by *old*. Reporting each range
separately duplicates table rows, and the name-keyed mappings derived
from the changes (compare URLs, release-notes items) would silently keep
one arbitrary range. Keeping the lowest parseable *old* per name yields a
single row whose compare URL and release notes subsume every narrower
range.
:param changes: `(name, old, new)` triples, possibly repeating a name.
:return: One triple per name, sorted, spanning from that name's lowest
*old* (`v` prefixes are ignored for the comparison).
"""
widest: dict[str, tuple[str, str]] = {}
for name, old, new in changes:
kept = widest.get(name)
if kept is None or is_newer(kept[0].removeprefix("v"), old.removeprefix("v")):
widest[name] = (old, new)
return [(name, old, new) for name, (old, new) in sorted(widest.items())]
def _plan_file_rewrites(
plan: SyncPlan,
rc: ResolveContext,
file_data: Mapping[Path, str],
rewriter: Callable[[str, Any], tuple[str, list[tuple[str, str, str]]]],
resolved: Any,
) -> None:
"""Apply a pin rewriter across every scanned file and record the outcome.
The write-planning half of the two `.github/` pin updaters: run the pure
rewriter over each file's text, keep the ones it changed for
{func}`_apply_file_writes`, and collapse the per-file change lists into one
row per name.
:param plan: The plan being resolved, mutated in place.
:param rc: The resolve inputs, supplying the cooldown.
:param file_data: Each scanned file's path and current text.
:param rewriter: A pure `(text, resolved) -> (new_text, changes)` function.
:param resolved: The rewriter's target versions, keyed however it expects.
"""
changes: list[tuple[str, str, str]] = []
for path, text in file_data.items():
new_content, file_changes = rewriter(text, resolved)
# Gate on the text, not on `file_changes`. A rewriter may edit a file
# without having a version move to report: `apply_workflow_literals`
# splices a missing self-pin cooldown exemption that way.
if new_content == text:
continue
plan.file_writes[path] = new_content
changes.extend(file_changes)
if not file_changes:
plan.self_pin_exemptions.append(path.name)
plan.rebase = lambda text: rewriter(text, resolved)
plan.changes = _widest_changes(changes)
plan.note_cooldown(rc.config.minimum_release_age, rc.min_age, rc.today)
def _action_releases(slug: str) -> list[Candidate]:
"""The GitHub releases of the action *slug*."""
return github_candidates(f"https://github.com/{slug}")
def _resolve_action_pins(rc: ResolveContext) -> SyncPlan:
"""Bump SHA-pinned GitHub Actions across `.github/` to their latest release.
An action pinned at more than one version also has its stragglers
converged onto the highest pin, even when no newer release clears the
cooldown, so every file agrees with the version the report claims.
"""
plan = SyncPlan(
operation="sync-action-pins",
subject="Action",
heading="Updated actions",
reference_date=rc.today,
)
file_data = rc.pinnable_files
# Highest currently-pinned version per slug, skipping repomatic-lineage refs
# (those thin-caller pins are managed by `repomatic init`). The winning
# pin's `(sha, ref)` is kept so slugs pinned at more than one version can
# converge their stragglers onto it without a network round-trip, even
# when no newer release clears the cooldown.
current: dict[str, str] = {}
current_pins: dict[str, tuple[str, str]] = {}
mixed: set[str] = set()
for text in file_data.values():
for pin in find_action_pins(text):
if pin.slug in UPSTREAM_REPO_SLUGS:
continue
version = pin.ref.removeprefix("v")
if pin.slug not in current:
current[pin.slug] = version
current_pins[pin.slug] = (pin.sha, pin.ref)
continue
if version != current[pin.slug]:
mixed.add(pin.slug)
if is_newer(version, current[pin.slug]):
current[pin.slug] = version
current_pins[pin.slug] = (pin.sha, pin.ref)
slugs = sorted(current)
releases = list(run_jobs(_action_releases, slugs, jobs=rc.jobs))
resolved: dict[str, tuple[str, str]] = {}
upgrades: dict[str, Candidate] = {}
for slug, candidates in zip(slugs, releases):
repo_url = f"https://github.com/{slug}"
current_version = current[slug]
latest = select_latest(candidates, rc.min_age, rc.today)
_track_held_back(plan, rc, slug, repo_url, candidates, latest, current_version)
if latest is not None and is_newer(latest.version, current_version):
upgrades[slug] = latest
continue
# No release beats the highest pin, but stragglers pinned at older
# versions still converge onto it: otherwise the repo-wide maximum
# masks them until a strictly newer release clears the cooldown, and
# the held-back table reports a "Locked" version that some files do
# not actually pin. The winning on-disk pin already carries the SHA,
# so no tag resolution is needed.
if slug in mixed:
resolved[slug] = current_pins[slug]
pinned_date = next(
(c.date for c in candidates if c.version == current_version), ""
)
if pinned_date:
plan.dates[slug] = pinned_date
# Resolving a tag to its commit is one more API call per upgrade: fan out.
upgrade_slugs = sorted(upgrades)
shas = run_jobs(
lambda slug: resolve_tag_to_sha(
f"https://github.com/{slug}", upgrades[slug].ref
),
upgrade_slugs,
jobs=rc.jobs,
)
for slug, new_sha in zip(upgrade_slugs, shas):
latest = upgrades[slug]
if not new_sha:
logging.warning(f"Could not resolve {slug}@{latest.ref} to a commit SHA.")
continue
resolved[slug] = (new_sha, latest.ref)
plan.dates[slug] = latest.date
_plan_file_rewrites(plan, rc, file_data, apply_action_pins, resolved)
plan.name_urls = {
slug: f"https://github.com/{slug}" for slug, _old, _new in plan.changes
}
plan.comparison_urls = {
slug: f"https://github.com/{slug}/compare/{old}...{new}"
for slug, old, new in plan.changes
}
if rc.release_notes:
# Actions tag as `vX.Y.Z`, so no per-tool extraction pattern is needed.
notes_items = [
ReleaseNotesRequest(
slug,
f"https://github.com/{slug}",
old.removeprefix("v"),
new.removeprefix("v"),
)
for slug, old, new in plan.changes
]
plan.notes_section = format_release_notes(
fetch_github_release_notes(notes_items)
)
return plan
def _gate_uv_on_checksums(
candidates: list[Candidate],
pinned: str,
file_data: Mapping[Path, str],
min_age: timedelta,
today: date,
) -> tuple[list[Candidate], bool]:
"""Drop uv releases the pinned `setup-uv` cannot checksum-verify.
The uv pin is the one literal whose adoptable range is decided by a
*second* pin: `setup-uv` verifies a download against a pinned hash only
for the versions in the checksum table its own release bundles, and gives
anything else no hash, or from `v10.1.0` one fetched at run time (see
{func}`~repomatic.deps.pins.setup_uv_verified_versions`).
Walking uv forward past that table therefore trades a pinned hash for a
live one at best, which is the opposite of what pinning it was for.
So the ceiling moves when the action pin moves, not when uv publishes. A
pin already sitting above that ceiling is the one case where the caller
walks a version backwards, down to the newest release the action can
verify. Waiting for `sync-action-pins` repairs nothing when the newest
`setup-uv` is the one already pinned, and every job keeps installing uv
without a pinned hash for as long as that holds.
An unreadable table holds the pin where it is. A bump past a table nobody
could read can overshoot it, and the first run that reads it again steps
the pin back: two pull requests that cancel out.
:param candidates: Every uv release, as offered by PyPI.
:param pinned: The uv version currently written in the workflows.
:param file_data: The scanned files, read for their `setup-uv` pins.
:param min_age: The stabilization window, to name what the gate withheld.
:param today: Reference date for the cooldown computation.
:return: The candidates carrying a checksum, paired with whether the pin on
disk is itself unverified. All of them pass when no `setup-uv` commit is
pinned, and only those no newer than *pinned* when a pinned table is
unreadable.
"""
shas = {
pin.sha
for text in file_data.values()
for pin in find_action_pins(text)
if pin.slug == SETUP_UV_SLUG
}
if not shas:
return candidates, False
verified = setup_uv_verified_versions(shas)
if verified is None:
logging.warning(
f"Holding the uv pin at {pinned} until the checksum table of the"
f" pinned {SETUP_UV_SLUG} can be read."
)
held = [
candidate
for candidate in candidates
if not is_newer(candidate.version, pinned)
]
return held, False
# Audit the pin already on disk, not just the one about to replace it: the
# same reason pin_inside_cooldown exists, for a condition that likewise
# entered the tree without passing this gate (hand-edited, or written
# before the gate existed). Reported here, repaired by the caller, which
# steps the pin back to the newest release this table covers.
unverified = pinned not in verified
if unverified:
logging.warning(
f"uv {pinned} is pinned but carries no checksum in the pinned"
f" {SETUP_UV_SLUG}, so no pinned hash verifies it in any job."
)
gated = [candidate for candidate in candidates if candidate.version in verified]
ungated_latest = select_latest(candidates, min_age, today)
gated_latest = select_latest(gated, min_age, today)
if ungated_latest is not None and (
gated_latest is None or is_newer(ungated_latest.version, gated_latest.version)
):
logging.info(
f"uv {ungated_latest.version} cleared the cooldown but carries no"
f" checksum in the pinned {SETUP_UV_SLUG}: holding the pin until"
" the action pin catches up."
)
return gated, unverified
def _docs_marker(marker: str) -> str:
"""Link a `sync-workflow-pins` table marker to the job's documentation.
:param marker: The emoji and words shown in the table cell.
:return: The marker as a link, or bare when the template names no docs.
"""
docs_url = PrTemplate.load("sync-workflow-pins").docs_url
return f"[{marker}]({docs_url})" if docs_url else marker
def _resolve_workflow_pins(rc: ResolveContext) -> SyncPlan:
"""Bump npm and PyPI version literals embedded in workflow YAML.
The upstream toolkit's inline pin (`repomatic==X.Y.Z`) is the exception:
it is aligned to the newest `uses:` ref version instead of the newest
cooldown-eligible PyPI release, in every shape
{func}`~repomatic.release.self_pin.inline_pin_re` matches, quoted or not.
The `lint-repo` lockstep check reads the same shapes, so it never flags a
pin this cannot fix. The refs are its source of truth: they are
written by `repomatic init` alone, which weighs the cooldown itself (see
{func}`~repomatic.init.upgrade.resolve_default_pin` and
{func}`~repomatic.init.upgrade.resolve_upgrade_target`) and is skipped here
and in `_resolve_action_pins`. Re-judging their outcome against PyPI would
leave `lint-repo`'s lockstep check red for a full cooldown window after
every refs bump.
"""
min_age, today = rc.min_age, rc.today
plan = SyncPlan(
operation="sync-workflow-pins",
subject="Package",
heading="Updated packages",
reference_date=today,
)
file_data = rc.pinnable_files
# Highest currently-pinned version per (ecosystem, package).
current: dict[tuple[Ecosystem, str], str] = {}
for text in file_data.values():
for literal in find_workflow_literals(text, self_pin=UPSTREAM_PACKAGE):
key = (literal.ecosystem, literal.package)
if key not in current or is_newer(literal.version, current[key]):
current[key] = literal.version
# Newest upstream `uses:` ref version, for the lockstep alignment.
lockstep_version: str | None = None
for text in file_data.values():
for version in find_upstream_ref_versions(text, DEFAULT_REPO):
if lockstep_version is None or is_newer(version, lockstep_version):
lockstep_version = version
resolved: dict[tuple[str, str], str] = {}
for (ecosystem, package), current_version in sorted(current.items()):
if (
ecosystem is Ecosystem.PYPI
and package == UPSTREAM_PACKAGE
and lockstep_version is not None
):
# Lockstep alignment, in either direction and regardless of the
# cooldown. Resolved unconditionally so every literal of the
# package realigns, even a straggler file lagging behind an
# already-aligned one; equal pins rewrite to themselves, which
# records no change. No held-back entry either: the pin tracks
# the refs, not PyPI, which also means no PyPI upload date is
# fetched: the "Released" cell marks the exemption instead.
resolved[(ecosystem, package)] = lockstep_version
plan.name_urls[package] = ecosystem.package_url(package)
plan.released_overrides[package] = _docs_marker(
"⛓️ lockstep with `uses:` refs"
)
continue
candidates = ecosystem.candidates(package)
unverified_uv_pin = False
if ecosystem is Ecosystem.PYPI and package == SETUP_UV_PACKAGE:
candidates, unverified_uv_pin = _gate_uv_on_checksums(
candidates, current_version, file_data, min_age, today
)
latest = select_latest(candidates, min_age, today)
# Audit the pin already on disk, not just the one about to replace it.
# A pin inside the window resolves through `uvx` in CI, which can read
# no per-package exemption, so it fails every job that installs it.
stuck = pin_inside_cooldown(candidates, current_version, min_age, today)
if stuck:
clears = format_countdown(
utc_midnight(stuck + min_age), utc_midnight(today)
)
logging.warning(
f"{package}=={current_version} is pinned inside the"
f" {rc.config.minimum_release_age} cooldown (published"
f" {stuck.isoformat()}). Installs resolving it from an index"
f" fail until it ages out on {clears}."
)
package_url = ecosystem.package_url(package)
_track_held_back(
plan, rc, package, package_url, candidates, latest, current_version
)
if latest is None:
continue
# The one move a pin makes downwards: a uv the pinned action cannot
# checksum is repaired by the newest release it can, which is older
# than what sits on disk. See _gate_uv_on_checksums.
step_back = unverified_uv_pin and is_newer(current_version, latest.version)
if step_back:
logging.warning(
f"Stepping the uv pin back from {current_version} to"
f" {latest.version}, the newest release the pinned"
f" {SETUP_UV_SLUG} can checksum-verify."
)
plan.change_labels[package] = _docs_marker("⏪ stepped back")
elif not is_newer(latest.version, current_version):
continue
resolved[(ecosystem, package)] = latest.version
plan.dates[package] = latest.date
plan.name_urls[package] = package_url
# The lockstep alignment above ignores the cooldown, so the pin it writes can
# name a release published minutes ago. Downstream workflows export a
# blanket `UV_EXCLUDE_NEWER` that `uvx` cannot override per package from the
# environment, so the exemption has to ride on the command line, exactly as
# the release freeze splices it into this repository's own workflows.
rewriter = partial(
apply_workflow_literals,
self_pin=UPSTREAM_PACKAGE,
)
_plan_file_rewrites(plan, rc, file_data, rewriter, resolved)
# A moved upstream pin is the one bump with a review to run behind it: the
# release notes list what changed, and the upgrade skill works them through
# against this repository. Rendered here rather than left to the template
# so the section exists only when the pin moved.
for name, old, new in plan.changes:
previous, adopted = old.removeprefix("v"), new.removeprefix("v")
if name == UPSTREAM_PACKAGE and is_newer(adopted, previous):
plan.upgrade_section = upgrade_invite(previous, adopted)
if rc.release_notes:
# Only PyPI literals resolve to a source repo, reusing sync-uv-lock's
# path: PyPI `project_urls` to GitHub releases, with a changelog-link
# fallback. npm literals have no discovery path, so they are left out.
pypi_packages = {pkg for eco, pkg in resolved if eco == Ecosystem.PYPI}
pypi_changes = [c for c in plan.changes if c[0] in pypi_packages]
if pypi_changes:
notes = fetch_release_notes(pypi_changes)
plan.notes_section = format_release_notes(notes)
return plan
def _apply_file_writes(plan: SyncPlan) -> None:
"""Rewrite each planned file from its current on-disk text.
Each file is re-read and the operation's rewriter replayed on it (see
{attr}`SyncPlan.rebase`), so a sibling operation's apply landing between
this plan's resolve and its write survives.
"""
if plan.rebase is None:
return
for path in plan.file_writes:
current = path.read_text(encoding="UTF-8")
rebased, _changes = plan.rebase(current)
write_if_changed(path, rebased, previous=current)
def _dep_sources_applies() -> bool:
"""`sync-dep-sources` runs wherever a git branch is tracked as a source."""
return bool(tracked_git_overrides(read_project(Path("pyproject.toml"))))
def _uv_lock_applies() -> bool:
"""`sync-uv-lock` runs wherever a `uv.lock` is present."""
return Path("uv.lock").is_file()
def _tool_versions_applies() -> bool:
"""`sync-tool-versions` runs only inside the repomatic source checkout.
It rewrites repomatic's own `tooling/registry.py`, so it is meaningful only when
that file lives under the current working tree (an editable checkout), never
from an installed wheel or a downstream repo.
"""
source = Path(tool_registry.__file__).resolve()
return Path.cwd().resolve() in source.parents
def _workflow_files_present() -> bool:
"""The pin updaters run wherever workflow or composite-action files exist."""
return bool(github_yaml_paths(Path()))
[docs]
def render_plan_markdown(plan: SyncPlan) -> str:
"""Render a plan as the markdown PR-body section every updater shares.
Concatenates the source-swap section (when the plan carries one), the
diff table, any release notes, the upgrade-review invitation, the uv
cooldown-bypass section, and the held-back section exactly as the
individual `sync-*` commands do, so `sync-deps` and the thin commands
produce identical output for the same plan. The invitation follows the
release notes because those are what it asks the maintainer to work
through.
Every other section reports what the run did to the working tree, so the
held-back one closes the body: it is the only forward-looking section,
listing releases the run deliberately left alone. A run that only rewrites
`exclude-newer-package` entries moves no version at all, and leading with
the forecast would open its PR on the releases it did not adopt instead of
the `pyproject.toml` hunk it asks to merge.
"""
swaps = plan.uv_project.source_swaps
swap_section = format_swap_section(
swaps,
name_urls=pypi_name_urls(swap.name for swap in swaps),
reference_date=plan.reference_date,
)
diff_table = format_diff_table(
plan.changes,
upload_times=plan.dates,
cooldown_note=plan.cooldown_note,
comparison_urls=plan.comparison_urls,
reference_date=plan.reference_date,
name_urls=plan.name_urls,
heading=plan.heading,
subject=plan.subject,
released_overrides=plan.released_overrides,
change_labels=plan.change_labels,
)
held_back_section = format_held_back_table(
plan.held_back,
plan.held_back_note,
name_urls=plan.held_back_name_urls,
subject=plan.subject,
)
# Bypasses are always PyPI packages (a uv-only concept), so their link
# targets are derived here rather than carried on the plan.
uv = plan.uv_project
bypass_section = format_bypass_section(
uv.bypass_forecasts,
pruned=uv.pruned_bypasses,
frozen=uv.frozen_bypasses,
stale_comments=uv.stale_bypass_comments,
name_urls=pypi_name_urls([
*(forecast.name for forecast in uv.bypass_forecasts),
*(forecast.name for forecast in uv.pruned_bypasses),
*uv.frozen_bypasses,
]),
)
# A splice-only run moves no version, so it reaches the diff table with
# nothing to put in it. Without this section its PR would carry a workflow
# hunk and no prose saying why one command grew a flag.
exemption_section = ""
if plan.self_pin_exemptions:
files = ", ".join(f"`{name}`" for name in sorted(plan.self_pin_exemptions))
exemption_section = (
"## 🩹 Restored cooldown exemption\n\n"
f"Spliced the missing `--exclude-newer-package` flag into {files}. The"
" inline pin moves in lockstep with the `uses:` refs, so it names a"
" release younger than the workflow-wide `UV_EXCLUDE_NEWER` window and"
" could not resolve at all without the exemption."
)
return "\n\n".join(
section
for section in (
swap_section,
diff_table,
plan.notes_section,
plan.upgrade_section,
bypass_section,
exemption_section,
held_back_section,
)
if section
)
[docs]
@dataclass(frozen=True)
class SyncOperation:
"""One cooldown-respecting dependency updater, as data.
The CLI command, workflow job step, PR branch and PR-body template all
share {attr}`name`, which makes a rename a one-line change.
"""
name: str
"""Command, branch, and template name (all identical)."""
config_key: str
"""The `[tool.repomatic]` key gating this operation, like `uv-lock.sync`."""
resolve: Callable[[ResolveContext], SyncPlan]
"""Read phase: network discovery, returns a {class}`SyncPlan`."""
apply: Callable[[SyncPlan], None]
"""Write phase: persist the plan's file writes."""
applies_here: Callable[[], bool]
"""Whether the operation is meaningful in the current working tree."""
[docs]
def is_enabled(self, config: Config) -> bool:
"""Whether this operation is enabled in *config*."""
return config.is_enabled(self.config_key)
SYNC_OPERATIONS: tuple[SyncOperation, ...] = (
SyncOperation(
name="sync-dep-sources",
config_key="dep-sources.sync",
resolve=_resolve_dep_sources,
apply=_apply_dep_sources,
applies_here=_dep_sources_applies,
),
SyncOperation(
name="sync-uv-lock",
config_key="uv-lock.sync",
resolve=_resolve_uv_lock,
apply=_apply_uv_lock,
applies_here=_uv_lock_applies,
),
SyncOperation(
name="sync-action-pins",
config_key="action-pins.sync",
resolve=_resolve_action_pins,
apply=_apply_file_writes,
applies_here=_workflow_files_present,
),
SyncOperation(
name="sync-workflow-pins",
config_key="workflow-pins.sync",
resolve=_resolve_workflow_pins,
apply=_apply_file_writes,
applies_here=_workflow_files_present,
),
SyncOperation(
name="sync-tool-versions",
config_key="tool-versions.sync",
resolve=_resolve_tool_versions,
apply=_apply_tool_versions,
applies_here=_tool_versions_applies,
),
)
"""The cooldown-respecting dependency updaters, in CI execution order.
`sync-dep-sources` first (adopting a release changes what the routine re-lock
even does), then `sync-uv-lock` (its lock churn gates other Python work), then
the two workflow-file rewriters, then the upstream-only tool bump last.
Only the first four share the `sync-deps` job in `autofix.yaml`. `sync-tool-versions`
runs from `self-maintenance.yaml` on its own daily schedule, since it rewrites this
package's source and has no downstream meaning; the order still applies to a local
`repomatic sync-deps`, which runs every enabled operation in one pass.
"""
OPERATIONS_BY_NAME: dict[str, SyncOperation] = {op.name: op for op in SYNC_OPERATIONS}
"""{data}`SYNC_OPERATIONS` keyed by {attr}`SyncOperation.name`."""
[docs]
def selected_operations(
config: Config, names: Sequence[str] | None = None
) -> list[SyncOperation]:
"""Return the operations to run, in {data}`SYNC_OPERATIONS` order.
The config feature flags are always authoritative: a disabled operation is
dropped whether or not it was named (mirrors each standalone `sync-*`
command, which exits when its flag is off).
:param config: The resolved configuration; disabled operations are dropped.
:param names: When given, restrict to these operation names. Unknown names
are ignored (the CLI validates them upstream). With none, operations
whose {attr}`SyncOperation.applies_here` is false (no `uv.lock`, no
workflow files, not the repomatic checkout) are dropped too: naming an
operation is an explicit opt-in that bypasses that probe (the "scope
exclusions are defaults, not absolutes" rule in `claude.md`).
"""
if names:
return [
op for op in SYNC_OPERATIONS if op.name in names and op.is_enabled(config)
]
return [op for op in SYNC_OPERATIONS if op.is_enabled(config) and op.applies_here()]
[docs]
def run_sync_operations(
operations: Sequence[SyncOperation],
rc: ResolveContext,
*,
spinner_label: str | None = None,
) -> list[tuple[SyncOperation, SyncPlan | None]]:
"""Resolve operations concurrently, then apply them serially.
The resolve phase fans out through {func}`click_extra.run_jobs` (the work is
network-bound and disjoint per operation), sized by {attr}`ResolveContext.jobs`.
A Ctrl+C drops queued resolves instead of waiting for them. When labelled,
an {class}`click_extra.OperationTrail` reports each resolve as a `✓`/`✘`
line and closes with a summary, its rendering tracking the worker count
and its elapsed times following `--time` (click-extra's own default). The
apply phase runs in {data}`SYNC_OPERATIONS` order because two of the five
rewrite the same workflow files. In `--dry-run` no apply runs. An operation
whose resolve raises is logged and reported with a `None` plan so one
failure never blocks the others.
:param operations: The operations to run (already filtered by the caller).
:param rc: Shared resolve inputs.
:param spinner_label: Present-tense label for the resolve trail (like
`"Resolving dependency updates"`). When set, the trail prints a
`✓`/`✘` line per operation on any stream, CI logs included, and a TTY
also gets a running tally. Unset (programmatic and test calls), the
trail is silent.
:return: Each operation paired with its plan (or `None` if its resolve
failed), in {data}`SYNC_OPERATIONS` order.
"""
if not operations:
return []
jobs = max(1, min(rc.jobs, len(operations)))
# A trail is opt-in: without a label (programmatic and test calls) it stays
# silent, so only the CLI's `spinner_label` lights it up.
trail = OperationTrail(
label=spinner_label or "",
unit="operations",
total=len(operations),
jobs=jobs,
visible=bool(spinner_label),
)
def resolve_and_mark(op: SyncOperation) -> SyncPlan | None:
try:
plan = op.resolve(rc)
except Exception:
logging.exception(f"{op.name} failed to resolve.")
plan = None
trail.mark(plan is not None, op.name)
return plan
with trail:
plans = {
op.name: plan
for op, plan in zip(
operations, run_jobs(resolve_and_mark, operations, jobs=jobs)
)
}
trail.finish(
trail.ok_count == len(operations),
f"Resolved {trail.ok_count}/{len(operations)} operations",
)
if not rc.dry_run:
for op in operations:
plan = plans.get(op.name)
if plan is not None and plan.has_changes:
op.apply(plan)
return [(op, plans.get(op.name)) for op in operations]