Lock/pin upstream versions for deterministic syncing

[Kilo59]

Summary

Pin upstream configs to a specific commit SHA for deterministic, reproducible syncing. Track the resolved version in a lockfile so check verifies against a known state, not just "latest".

Parent: #100 (Tier 1)

Motivation

Without pinning, ruff-sync pull in CI could silently pick up breaking upstream changes. Every major package ecosystem has a lockfile mechanism:

Ecosystem	Lock Mechanism
npm	package-lock.json records exact resolved versions
Go	go.sum records content hashes
pre-commit	rev: <sha/tag> pins exact hook versions
pip	pip freeze / uv lock

ruff-sync should let users opt into deterministic syncing while keeping the default behavior unchanged.

Proposed Design

Lock storage

Store lock metadata in pyproject.toml under [tool.ruff-sync.lock] rather than a separate file. This keeps everything in one place and avoids adding a new file to every project.

# Written by ruff-sync after a successful pull
[tool.ruff-sync.lock]
upstream = "https://raw.githubusercontent.com/my-org/standards/main/pyproject.toml"
commit = "abc1234def5678..."           # resolved commit SHA (if git-resolvable)
content-hash = "sha256:e3b0c44..."     # hash of the upstream ruff config section
pulled-at = "2026-03-15T15:30:00Z"     # timestamp of last pull

New CLI commands / flags

# Pull and update lock (default behavior when lock section exists)
ruff-sync pull                      # fetches latest, updates lock

# Pull but skip lock update (useful for testing)
ruff-sync pull --no-lock

# Check against locked version specifically
ruff-sync check                     # uses lock hash if available

# Explicitly update lock without applying changes
ruff-sync lock                      # new subcommand: fetch, resolve, write lock only

Implementation Plan

1. Define lock schema in Config TypedDict (core.py)

class LockInfo(TypedDict, total=False):
    """Lock metadata written after a successful pull."""
    upstream: str          # resolved raw URL
    commit: str            # git commit SHA if resolvable
    content_hash: str      # sha256 of the upstream ruff config text
    pulled_at: str         # ISO 8601 timestamp

2. Compute content hash (core.py)

import hashlib

def compute_config_hash(config_text: str) -> str:
    """Compute a deterministic hash of the upstream ruff config."""
    # Normalize: parse and re-serialize to ignore whitespace variance
    normalized = tomlkit.dumps(tomlkit.parse(config_text))
    return f"sha256:{hashlib.sha256(normalized.encode()).hexdigest()}"

3. Resolve commit SHA (core.py)

For GitHub/GitLab URLs, resolve the current commit SHA via API or from the git clone:

async def resolve_commit_sha(
    url: URL, branch: str, client: httpx.AsyncClient
) -> str | None:
    """Resolve the current commit SHA for a GitHub/GitLab branch."""
    # GitHub API: GET /repos/{owner}/{repo}/commits/{branch}
    if url.host in _GITHUB_HOSTS or url.host == _GITHUB_RAW_HOST:
        # Extract org/repo from URL
        ...
        api_url = f"https://api.github.com/repos/{org}/{repo}/commits/{branch}"
        resp = await client.get(
            api_url, headers={"Accept": "application/vnd.github.sha"}
        )
        if resp.status_code == 200:
            return resp.text.strip()
    return None  # fallback: no SHA available

For git clone fetches, extract SHA from the cloned repo.

4. Write lock after pull() (core.py)

After a successful merge, write lock metadata to pyproject.toml:

async def pull(args: Arguments) -> int:
    # ... existing fetch + merge logic ...

    # Write lock metadata (only for pyproject.toml targets)
    if not args.no_lock and _source_toml_path.name == "pyproject.toml":
        lock_info = {
            "upstream": str(fetch_result.resolved_upstream),
            "content-hash": compute_config_hash(upstream_config_text),
            "pulled-at": dt.datetime.now(dt.timezone.utc).isoformat(),
        }
        if commit_sha:
            lock_info["commit"] = commit_sha
        # Write into [tool.ruff-sync.lock] using tomlkit
        _write_lock(merged_toml, lock_info)
    source_toml_file.write(merged_toml)

5. Check against lock in check() (core.py)

If lock metadata exists, compare the upstream content hash against the locked hash for a fast "has upstream changed?" check:

async def check(args: Arguments) -> int:
    # ... existing logic ...

    # If lock exists, also verify upstream hasn't changed since last pull
    config = get_config(args.to)
    if "lock" in config:
        lock = config["lock"]
        upstream_hash = compute_config_hash(upstream_config_text)
        if upstream_hash != lock.get("content-hash"):
            print("Warning: Upstream has changed since last pull "
                  f"(locked at {lock.get('pulled-at', 'unknown')})")

6. Add lock subcommand (cli.py)

A lightweight subcommand that fetches upstream, resolves the commit SHA and content hash, and writes the lock section without modifying the ruff config:

lock_parser = subparsers.add_parser(
    "lock",
    parents=[common_parser],
    help="Fetch upstream and update lock metadata without changing ruff config",
)

7. Add --no-lock flag to pull (cli.py)

pull_parser.add_argument(
    "--no-lock",
    action="store_true",
    help="Skip updating the lock metadata after pull.",
)

Backward Compatibility

• Lock is opt-in: if no [tool.ruff-sync.lock] section exists, behavior is unchanged.
• The lock section is written automatically on the first pull (can be disabled with --no-lock).
• check gracefully handles missing lock sections.
• Lock metadata is stored using tomlkit to preserve formatting of the rest of pyproject.toml.

Edge Cases

• ruff.toml targets: Lock metadata can't go in ruff.toml (no [tool] section). Options: (a) skip locking, (b) store in a nearby pyproject.toml, or (c) use a standalone .ruff-sync.lock file. Recommend (a) for MVP.
• Git clone upstreams: SHA is directly available from the clone; no API call needed.
• Non-GitHub/GitLab hosts: Content hash still works; commit SHA may not be resolvable (logged as info, not an error).

Test Plan

1. Unit test for compute_config_hash() — verify deterministic hashing.
2. Unit test for _write_lock() — verify tomlkit preserves formatting when adding lock section.
3. E2E test — pull writes lock, subsequent check passes, modify upstream, check reports both hash mismatch and content drift.
4. --no-lock test — verify lock section is not written.
5. lock subcommand test — verify it writes lock without modifying ruff config.
6. Missing lock test — verify check works normally without lock section.

Files Changed

File	Change
src/ruff_sync/core.py	LockInfo TypedDict, compute_config_hash(), resolve_commit_sha(), _write_lock(), update pull() and check()
src/ruff_sync/cli.py	lock subcommand, --no-lock flag, Arguments.no_lock field
src/ruff_sync/__init__.py	Export LockInfo
tests/test_basic.py	Hash and lock-write unit tests
tests/test_e2e.py	E2E lock lifecycle tests

[Kilo59]

Architectural Overview

The goal is to decouple Sync Logic (calculating what needs to be updated) from Storage Logic (how and where that data is saved). This prevents pyproject.toml from being the only possible storage location, solving the Docker cache invalidation issue (among other potential issues).

---

1. Data Models (ruff_sync/models.py)

Define structured objects for the lock data. [cite_start]This ensures the rest of the application works with typed objects instead of raw dictionaries[cite: 86, 99].

from dataclasses import dataclass, field, asdict
from typing import List, Optional

@dataclass
class LockLayer:
    """
    Metadata for a single configuration source.
    Decision: Storing individual layers allows ruff-sync to verify specific 
    [cite_start]upstreams without re-fetching everything if only one changes[cite: 77, 107].
    """
    upstream_url: str
    resolved_hash: str
    timestamp: str

@dataclass
class LockState:
    """
    The full state of the project's locked configuration.
    [cite_start]Decision: Including the tool version ensures future compatibility checks[cite: 51, 100].
    """
    ruff_sync_version: str = "0.1.0"
    layers: List[LockLayer] = field(default_factory=list)

    def to_dict(self) -> dict:
        """Convert to a dictionary suitable for TOML serialization."""
        return asdict(self)

2. The Storage Protocol (ruff_sync/lock_store.py)

[cite_start]Using a Protocol allows for structural typing. Any class implementing these three methods is valid, meaning the core logic never needs to know which file it is talking to[cite: 97, 100, 105].

from typing import Protocol, Optional
from pathlib import Path

class LockStore(Protocol):
    """
    The interface for reading and writing lock data.
    """
    def read(self) -> Optional[LockState]:
        [cite_start]"""Load lock data. Returns None if no lock exists[cite: 65, 100]."""
        ...

    def write(self, state: LockState) -> None:
        [cite_start]"""Persist the LockState to the preferred storage[cite: 66, 100]."""
        ...

    def exists(self) -> bool:
        [cite_start]"""Check for the presence of a lock file/section[cite: 66, 100]."""
        ...

3. Implementation: EmbeddedLockStore

[cite_start]This class handles the initial requirement of storing data in pyproject.toml. It uses tomlkit to preserve "artisanal" user formatting[cite: 7, 90, 102].

import tomlkit
from pathlib import Path

class EmbeddedLockStore:
    def __init__(self, pyproject_path: Path):
        self.path = pyproject_path

    def read(self) -> Optional[LockState]:
        if not self.path.exists():
            return None
            
        doc = tomlkit.parse(self.path.read_text())
        # [cite_start]Path: [tool.ruff-sync.lock] [cite: 39, 69]
        data = doc.get("tool", {}).get("ruff-sync", {}).get("lock")
        
        if not data:
            return None
            
        return LockState(
            ruff_sync_version=data.get("ruff_sync_version", "0.1.0"),
            layers=[LockLayer(**l) for l in data.get("layers", [])]
        )

    def write(self, state: LockState) -> None:
        # [cite_start]Load existing file to preserve comments/formatting [cite: 7, 106]
        doc = tomlkit.parse(self.path.read_text()) if self.path.exists() else tomlkit.document()
        
        # [cite_start]Build nested table safely [cite: 8]
        tool = doc.setdefault("tool", tomlkit.table())
        ruff_sync = tool.setdefault("ruff-sync", tomlkit.table())
        
        # Replace only the lock table
        ruff_sync["lock"] = state.to_dict()
        
        self.path.write_text(tomlkit.dumps(doc))

    def exists(self) -> bool:
        if not self.path.exists():
            return False
        doc = tomlkit.parse(self.path.read_text())
        return "lock" in doc.get("tool", {}).get("ruff-sync", {})

4. The Resolver (ruff_sync/lock_store.py)

This is the "brain" that decides which store to use. [cite_start]It implements the "Detect and Act" strategy[cite: 52, 71, 103].

class LockStoreResolver:
    def __init__(self, project_root: Path):
        self.root = project_root
        self.pyproject = project_root / "pyproject.toml"
        self.standalone = project_root / "ruff-sync.lock"

    def get_store(self) -> LockStore:
        """
        Decision: Prioritize the standalone lock file.
        If a user manually creates 'ruff-sync.lock', the tool will 
        immediately start using it, preventing pyproject.toml changes 
        [cite_start]from busting Docker caches[cite: 29, 74, 94].
        """
        if self.standalone.exists():
            # Return StandaloneLockStore(self.standalone) when implemented
            pass 
            
        return EmbeddedLockStore(self.pyproject)

5. Integration into Core Logic (ruff_sync/core.py)

[cite_start]The pull and check functions should no longer touch pyproject.toml for locking. They should receive a LockStore via dependency injection[cite: 80, 87, 91].

def pull(config: dict, store: LockStore):
    """
    Example of how the agent should use the store.
    """
    # [cite_start]1. Read existing lock [cite: 65, 100]
    current_lock = store.read()
    
    # [cite_start]2. Logic to fetch upstreams and generate new LockState [cite: 20, 26]
    new_layers = [] 
    # ... fetch logic ...
    
    new_state = LockState(layers=new_layers)
    
    # [cite_start]3. Write new lock [cite: 66, 100]
    store.write(new_state)

Final Instructions for the Agent

1. Strict File Boundaries: Do not put TOML parsing logic inside core.py. [cite_start]Keep it inside the LockStore implementations[cite: 79, 83].
2. [cite_start]Preserve Formatting: Always use tomlkit for EmbeddedLockStore to ensure local overrides and comments are not lost[cite: 7, 102, 106].
3. [cite_start]Future-Proofing: If a user runs a command, always check for ruff-sync.lock first via the Resolver[cite: 72, 104].

Does this level of detail provide enough structure for your implementation, or would you like to see the specific logic for ruff-sync check using these new protocols?

[Kilo59]

🛠 Implementation Checklist: ruff-sync Lock File

Phase 1: Data Models (ruff_sync/models.py)

• [cite_start]Create LockLayer dataclass: Define metadata for single configuration sources, including upstream_url, resolved_hash, and timestamp. [cite: 158]
• [cite_start]Create LockState dataclass: Standardize the project's full locked state, including the tool version and a list of LockLayer objects. [cite: 159]
• [cite_start]Implement serialization: Add a to_dict() method to the models for easy TOML conversion. [cite: 160]

Phase 2: The Interface & Storage (ruff_sync/lock_store.py)

• [cite_start]Define LockStore Protocol: Establish the required behavior with read(), write(), and exists() methods to keep the core logic storage-agnostic. [cite: 161]
• [cite_start]Implement EmbeddedLockStore: Create a concrete class that targets the [tool.ruff-sync.lock] table within pyproject.toml. [cite: 162]
• [cite_start]Integrate tomlkit: Ensure EmbeddedLockStore uses tomlkit to preserve "artisanal" user comments and formatting when updating hashes. [cite: 163]
• [cite_start]Implement LockStoreResolver: Build the resolution logic to prioritize ruff-sync.lock over pyproject.toml. [cite: 164]

Phase 3: Core & CLI Integration

• [cite_start]Inject Dependencies in Core: Refactor core.pull to accept a LockStore instance rather than opening files directly. [cite: 165]
• [cite_start]Update Check Logic: Modify core.check to use the injected LockStore for drift detection across any storage format. [cite: 166]
• [cite_start]Enforce File Boundaries: Remove all TOML parsing logic from core.py and move it into the appropriate LockStore implementations. [cite: 167]
• [cite_start]Wire CLI Entry Points: Update CLI commands to use LockStoreResolver to fetch the correct store and pass it to core functions. [cite: 168]

Phase 4: Final Polishing

• [cite_start]Address Migration Gaps: Decide if the tool should automatically import state from pyproject.toml when a new ruff-sync.lock is detected. [cite: 169]
• [cite_start]Verify CI Compatibility: Ensure ruff-sync check correctly identifies drift in both embedded and standalone scenarios. [cite: 170]

---

Design Summary: The Storage-Agnostic Bridge

[cite_start]This design utilizes a Storage-Agnostic Bridge to decouple your synchronization logic from the filesystem[cite: 155]. [cite_start]By using a Protocol for structural subtyping instead of rigid inheritance, ruff-sync can interact with either an embedded pyproject.toml section or a standalone ruff-sync.lock interchangeably without the core logic knowing the difference[cite: 156, 131].

[cite_start]This "Detect and Act" approach prioritizes standalone files to solve the Docker cache issue: if a ruff-sync.lock exists, the tool uses it exclusively, keeping pyproject.toml static and preventing unnecessary layer invalidation during builds[cite: 157, 132, 118].