From 8c406c87c4ff7beac012efd472af717f5952266a Mon Sep 17 00:00:00 2001 From: Brian Krabach Date: Thu, 1 Oct 2026 04:42:34 -0700 Subject: [PATCH 1/3] Add opt-in shared Git sources and preparation receipts --- amplifier_foundation/modules/activator.py | 38 +- amplifier_foundation/modules/preparation.py | 104 ++++++ amplifier_foundation/sources/git.py | 60 +++- amplifier_foundation/sources/shared.py | 345 +++++++++++++++++++ docs/shared-source-store.md | 36 ++ tests/test_dependency_refresh.py | 65 ++++ tests/test_dependency_refresh_integration.py | 48 +++ tests/test_shared_sources.py | 216 ++++++++++++ 8 files changed, 903 insertions(+), 9 deletions(-) create mode 100644 amplifier_foundation/modules/preparation.py create mode 100644 amplifier_foundation/sources/shared.py create mode 100644 docs/shared-source-store.md create mode 100644 tests/test_shared_sources.py diff --git a/amplifier_foundation/modules/activator.py b/amplifier_foundation/modules/activator.py index 112f8bf6..a0a1d300 100644 --- a/amplifier_foundation/modules/activator.py +++ b/amplifier_foundation/modules/activator.py @@ -574,6 +574,23 @@ async def _install_dependencies( module_name: str | None = None, progress_callback: Callable[[str, str], None] | None = None, force: bool = False, + ) -> None: + from .preparation import PreparationReceipt + + receipt = PreparationReceipt(self, module_path) + if not force and await asyncio.to_thread(receipt.reusable): + return + await self._install_dependencies_uncached( + module_path, module_name, progress_callback, force + ) + await asyncio.to_thread(receipt.record) + + async def _install_dependencies_uncached( + self, + module_path: Path, + module_name: str | None = None, + progress_callback: Callable[[str, str], None] | None = None, + force: bool = False, ) -> None: """Install Python dependencies for a module. @@ -675,6 +692,13 @@ async def _install_dependencies( requirements = module_path / "requirements.txt" if pyproject.exists(): + from amplifier_foundation.sources.shared import build_lock, build_view + + install_path, immutable = await asyncio.to_thread(build_view, module_path) + # Shared source trees are read-only. Build wheels from a private + # writable input; never create an editable installation on them. + # uv reuses the wheel for the same source and build contract. + # Build metadata may change without rewriting the canonical source. # Build overrides for git URL dependencies that are already installed. # This prevents uv from fetching/building packages from git when a # prebuilt wheel is already available (e.g. amplifier-core from PyPI). @@ -689,8 +713,8 @@ async def _install_dependencies( "uv", "pip", "install", - "-e", - str(module_path), + *([] if immutable else ["-e"]), + str(install_path), "--python", self.install_python, "--quiet", @@ -724,12 +748,10 @@ async def _install_dependencies( overrides_file.close() cmd.extend(["--overrides", overrides_file.name]) - subprocess.run( - cmd, - check=True, - capture_output=True, - text=True, - ) + # Holding this across uv prevents concurrent build backends + # from modifying the same writable source view. + with build_lock(install_path): + subprocess.run(cmd, check=True, capture_output=True, text=True) # Mark as installed after successful install self._install_state.mark_installed(module_path) # Refresh Python's package discovery so subprocess-installed packages diff --git a/amplifier_foundation/modules/preparation.py b/amplifier_foundation/modules/preparation.py new file mode 100644 index 00000000..3a4fc0a9 --- /dev/null +++ b/amplifier_foundation/modules/preparation.py @@ -0,0 +1,104 @@ +"""Reuse a completed install only inside one isolated preparation attempt. + +Every new attempt refreshes each selected source at least once. A duplicate +may skip resolution only when the source, explicit policy and complete installed +package metadata still match. Any intervening graph change forces a fresh install. +This is not a standing installed-version shortcut or a dependency pin. +""" + +from __future__ import annotations + +import hashlib +import importlib.metadata +import json +import os +import re +import sys +from pathlib import Path + + +def graph_signature(): + rows = [] + for dist in importlib.metadata.distributions(): + rows.append( + [ + dist.metadata["Name"], + dist.version, + dist.read_text("METADATA"), + dist.read_text("direct_url.json"), + ] + ) + return hashlib.sha256(json.dumps(sorted(rows), sort_keys=True).encode()).hexdigest() + + +def source_signature(root): + root = Path(root) + digest = hashlib.sha256() + for path in sorted(root.rglob("*")): + relative = path.relative_to(root) + if any( + part in {".git", ".venv", "node_modules", "__pycache__", "build", "dist"} + or part.endswith(".egg-info") + for part in relative.parts + ): + continue + if path.is_symlink(): + if not path.resolve().is_relative_to(root.resolve()): + # External build inputs can change independently. Unknown + # content must force installation, not a reusable receipt. + raise ValueError("External source symlink cannot be qualified") + digest.update(str(relative).encode() + b"\0" + os.readlink(path).encode()) + elif path.is_file(): + digest.update(str(relative).encode() + b"\0" + path.read_bytes()) + return digest.hexdigest() + + +class PreparationReceipt: + def __init__(self, activator, source): + self.source = Path(source) + token = os.environ.get("AMPLIFIER_INSTALL_PREPARATION", "") + # An external install target cannot be qualified by this process's graph. + self.path = None + if ( + activator.refresh_dependencies + and re.fullmatch("[a-f0-9]{32}", token) + and os.path.abspath(activator.install_python) + == os.path.abspath(sys.executable) + ): + policy = [ + Path(value).read_bytes().hex() if value else None + for value in ( + activator.install_constraints, + activator.install_overrides, + ) + ] + self.policy = hashlib.sha256(json.dumps(policy).encode()).hexdigest() + key = hashlib.sha256(str(self.source.absolute()).encode()).hexdigest() + self.path = ( + activator.cache_dir / "install-preparations" / token / (key + ".json") + ) + + def evidence(self): + return { + "source": source_signature(self.source), + "policy": self.policy, + "graph": graph_signature(), + } + + def reusable(self): + if not self.path: + return False + try: + return json.loads(self.path.read_text()) == self.evidence() + except (OSError, ValueError, TypeError): + return False + + def record(self): + if self.path: + from amplifier_foundation.settings import atomic_write + + try: + self.path.parent.mkdir(parents=True, exist_ok=True) + atomic_write(self.path, json.dumps(self.evidence()), private=True) + except (OSError, ValueError, TypeError): + pass # Reuse is optional; later qualification remains authoritative. diff --git a/amplifier_foundation/sources/git.py b/amplifier_foundation/sources/git.py index 40ae93cd..709d98b0 100644 --- a/amplifier_foundation/sources/git.py +++ b/amplifier_foundation/sources/git.py @@ -411,6 +411,14 @@ def _build_git_url(self, parsed: ParsedURI) -> str: scheme = parsed.scheme.replace("git+", "") return f"{scheme}://{parsed.host}{parsed.path}" + def _shared_uri(self, parsed): + uri = "git+" + self._build_git_url(parsed) + if parsed.ref: + uri += "@" + parsed.ref + if parsed.subpath: + uri += "#subdirectory=" + parsed.subpath + return uri + def _get_cache_path(self, parsed: ParsedURI, cache_dir: Path) -> Path: """Get the cache path for a parsed URI.""" git_url = self._build_git_url(parsed) @@ -672,6 +680,20 @@ async def resolve(self, parsed: ParsedURI, cache_dir: Path) -> ResolvedSource: Raises: BundleNotFoundError: If clone fails or ref not found. """ + if os.environ.get("AMPLIFIER_SOURCE_STORE"): + from .shared import resolve_shared_source + + result = await resolve_shared_source( + parsed.original + if hasattr(parsed, "original") + else self._shared_uri(parsed), + cache_dir, + ) + if not self._verify_clone_integrity(result.source_root): + raise BundleNotFoundError( + "Shared source is not a bundle or module repository" + ) + return result cache_path = self._get_cache_path(parsed, cache_dir) cache_path.parent.mkdir(parents=True, exist_ok=True) # Workers share this cache across processes. Check integrity only after @@ -829,6 +851,31 @@ async def get_status(self, parsed: ParsedURI, cache_dir: Path) -> SourceStatus: source_uri += f"#subdirectory={parsed.subpath}" # Initialize status + if shared_root := os.environ.get("AMPLIFIER_SOURCE_STORE"): + from .shared import SharedSourceStore + + try: + shared_path = await asyncio.to_thread( + SharedSourceStore(shared_root).cached, source_uri, cache_dir + ) + if shared_path is not None: + cache_path = shared_path + except ( + OSError, + ValueError, + KeyError, + TypeError, + subprocess.SubprocessError, + ): + return SourceStatus( + source_uri=source_uri, + is_cached=False, + cached_ref=ref, + remote_ref=ref, + has_update=None, + error="Shared source could not be verified", + summary="Shared source was retained for inspection", + ) status = SourceStatus( source_uri=source_uri, is_cached=cache_path.exists(), @@ -856,7 +903,12 @@ async def get_status(self, parsed: ParsedURI, cache_dir: Path) -> SourceStatus: # Get remote commit try: - status.remote_commit = await self._get_remote_commit(git_url, ref) + if shared_root: + from .shared import remote_revision + + status.remote_commit = await remote_revision(git_url, ref) + else: + status.remote_commit = await self._get_remote_commit(git_url, ref) if status.remote_commit is None: status.has_update = None @@ -900,6 +952,12 @@ async def update(self, parsed: ParsedURI, cache_dir: Path) -> ResolvedSource: Raises: BundleNotFoundError: If clone fails. """ + if os.environ.get("AMPLIFIER_SOURCE_STORE"): + from .shared import resolve_shared_source + + return await resolve_shared_source( + self._shared_uri(parsed), cache_dir, refresh=True + ) cache_path = self._get_cache_path(parsed, cache_dir) cache_path.parent.mkdir(parents=True, exist_ok=True) async with AsyncFileLock(cache_path.with_name(f".{cache_path.name}.lock")): diff --git a/amplifier_foundation/sources/shared.py b/amplifier_foundation/sources/shared.py new file mode 100644 index 00000000..32ca37a0 --- /dev/null +++ b/amplifier_foundation/sources/shared.py @@ -0,0 +1,345 @@ +"""Opt-in, commit-addressed sources shared by bundles, modules and skills. + +Applications own branch bindings. The store owns immutable checkouts. Refresh +creates a new binding; it never checks out files beneath an existing reader. +Absent AMPLIFIER_SOURCE_STORE, existing resolver behavior is unchanged. +""" + +from __future__ import annotations + +import asyncio +import hashlib +import json +import os +import re +import shutil +import stat +import subprocess +import tempfile +from pathlib import Path +from urllib.parse import urlsplit + +from filelock import AsyncFileLock + +from amplifier_foundation.paths.resolution import ResolvedSource, parse_uri + + +async def complete_io(function, *args, **kwargs): + """Do not release a source writer lock while a cancelled thread is writing.""" + task = asyncio.create_task(asyncio.to_thread(function, *args, **kwargs)) + try: + return await asyncio.shield(task) + except asyncio.CancelledError: + await task + raise + + +def binding_root(cache): + cache = Path(cache) + return cache.parent if cache.name in {"skills", "bundles"} else cache + + +def source_id(url, ref): + parsed = urlsplit(url) + if parsed.username or parsed.password or parsed.query or parsed.fragment: + raise ValueError("Shared sources require credential-free repository URLs") + url = parsed._replace(path=parsed.path.rstrip("/").removesuffix(".git")).geturl() + return hashlib.sha256((url + "@" + ref).encode()).hexdigest() + + +def atomic(path, value): + from amplifier_foundation.settings import atomic_write + + path.parent.mkdir(parents=True, exist_ok=True) + atomic_write(path, json.dumps(value), private=True) + + +def clean_checkout(root): + """Tracked changes and unknown extras are local work, even when ignored.""" + root = Path(root) + if not (root / ".git").is_dir() or (root / ".git").is_symlink(): + return False # Linked worktrees can refer to a mutable external Git dir. + dirty = subprocess.check_output( + ["git", "--no-optional-locks", "status", "--porcelain", "--untracked-files=no"], + cwd=root, + text=True, + ) + extras = subprocess.check_output( + ["git", "ls-files", "--others", "-z"], cwd=root + ).split(b"\0") + extras = [name for name in extras if name and name != b".amplifier_cache_meta.json"] + return not dirty and not extras + + +class SharedSourceStore: + def __init__(self, root): + self.root = Path(root).expanduser().resolve() + + def checkout(self, url, revision): + if not re.fullmatch(r"(?:[a-f0-9]{40}|[a-f0-9]{64})", revision): + raise ValueError("An exact source revision is required") + target = self.root / "objects" / source_id(url, "") / revision + if target.is_symlink() or not target.resolve().is_relative_to(self.root): + raise ValueError("Shared source object is outside its owned store") + return target + + def verify(self, url, revision): + target = self.checkout(url, revision) + metadata = json.loads((target / ".amplifier_cache_meta.json").read_text()) + if ( + source_id(metadata.get("git_url", ""), "") != source_id(url, "") + or metadata.get("commit") != revision + ): + raise ValueError("Shared source identity changed") + actual = subprocess.check_output( + ["git", "rev-parse", "HEAD"], cwd=target, text=True + ).strip() + if actual != revision or not clean_checkout(target): + raise ValueError("Shared source contents changed") + return target + + async def ensure(self, url, revision, *, existing=None): + target = self.checkout(url, revision) + target.parent.mkdir(parents=True, exist_ok=True) + async with AsyncFileLock(target.with_name("." + revision + ".lock")): + if target.exists(): + return await asyncio.to_thread(self.verify, url, revision) + stage = Path(tempfile.mkdtemp(prefix=".source-", dir=target.parent)) + try: + from .git import GitSourceHandler + + if existing is not None: + if not await asyncio.to_thread(clean_checkout, existing): + raise ValueError( + "Only exact, clean sources may enter the shared store" + ) + await complete_io( + shutil.copytree, + existing, + stage, + dirs_exist_ok=True, + symlinks=True, + ) + else: + await complete_io( + GitSourceHandler()._clone_at_commit, url, revision, stage + ) + + def publish(): + atomic( + stage / ".amplifier_cache_meta.json", + { + "git_url": url, + "ref": revision, + "commit": revision, + "immutable": True, + }, + ) + actual = subprocess.check_output( + ["git", "rev-parse", "HEAD"], cwd=stage, text=True + ).strip() + dirty = subprocess.check_output( + [ + "git", + "--no-optional-locks", + "status", + "--porcelain", + "--untracked-files=no", + ], + cwd=stage, + text=True, + ) + if actual != revision or dirty or not clean_checkout(stage): + raise ValueError( + "Only exact, clean sources may enter the shared store" + ) + # Builds must use writable build views, never modify this tree. + for path in stage.rglob("*"): + if not path.is_symlink(): + path.chmod(stat.S_IMODE(path.stat().st_mode) & ~0o222) + stage.chmod(0o555) + stage.rename(target) + + await complete_io(publish) + finally: + if stage.exists(): + from ._rmtree import rmtree_robust + + await complete_io(rmtree_robust, stage) + return target + + async def bind(self, cache, url, ref, revision, *, existing=None): + target = await self.ensure(url, revision, existing=existing) + path = ( + binding_root(cache) / ".source-bindings" / (source_id(url, ref) + ".json") + ) + atomic(path, {"git_url": url, "ref": ref, "commit": revision}) + return target + + def cached(self, uri, cache): + """Inspect cached status without creating a binding or cloning.""" + from .git import GitSourceHandler + + parsed = parse_uri(uri) + handler = GitSourceHandler() + url, ref = handler._build_git_url(parsed), parsed.ref or "HEAD" + path = ( + binding_root(cache) / ".source-bindings" / (source_id(url, ref) + ".json") + ) + legacy = handler._get_cache_path(parsed, Path(cache)) + if legacy.exists() and (not path.exists() or not clean_checkout(legacy)): + return legacy + if not path.exists(): + return None + record = json.loads(path.read_text()) + if source_id(record["git_url"], record["ref"]) != source_id(url, ref): + raise ValueError("Shared source binding changed") + return self.verify(url, record["commit"]) + + async def resolve(self, uri, cache, *, refresh=False): + parsed = parse_uri(uri) + from .git import GitSourceHandler + + handler = GitSourceHandler() + url, ref = handler._build_git_url(parsed), parsed.ref or "HEAD" + path = ( + binding_root(cache) / ".source-bindings" / (source_id(url, ref) + ".json") + ) + path.parent.mkdir(parents=True, exist_ok=True) + async with AsyncFileLock(path.with_suffix(".lock")): + # Binding scopes coalesce bundles/skills, but legacy directories + # retain the handler's original cache path and any edits there. + legacy = handler._get_cache_path(parsed, Path(cache)) + if ( + legacy.exists() + and ( + not path.exists() + or not await asyncio.to_thread(clean_checkout, legacy) + ) + and not refresh + ): + active = legacy / parsed.subpath if parsed.subpath else legacy + if ( + not active.resolve().is_relative_to(legacy.resolve()) + or not active.exists() + ): + raise ValueError("Shared source subpath is unavailable") + return ResolvedSource(active_path=active, source_root=legacy) + if ( + refresh + and legacy.exists() + and not await asyncio.to_thread(clean_checkout, legacy) + ): + raise ValueError( + "Local changes retained; refresh requires a clean source" + ) + if path.exists() and not refresh: + record = json.loads(path.read_text()) + if source_id(record["git_url"], record["ref"]) != source_id(url, ref): + raise ValueError("Shared source binding changed") + target = await self.ensure(url, record["commit"]) + else: + revision = ( + ref + if re.fullmatch(r"(?:[a-f0-9]{40}|[a-f0-9]{64})", ref) + else await remote_revision(url, ref) + ) + if not revision: + raise ValueError("Shared source ref is unavailable") + target = await self.bind(cache, url, ref, revision) + active = target / parsed.subpath if parsed.subpath else target + if not active.resolve().is_relative_to(target) or not active.exists(): + raise ValueError("Shared source subpath is unavailable") + return ResolvedSource(active_path=active, source_root=target) + + +async def remote_revision(url, ref): + """Resolve explicit branch/tag names without selecting an ambiguous match.""" + patterns = ( + [ref] + if ref == "HEAD" or ref.startswith("refs/") + else ["refs/heads/" + ref, "refs/tags/" + ref, "refs/tags/" + ref + "^{}"] + ) + + def read(): + output = subprocess.check_output( + ["git", "ls-remote", url, *patterns], text=True, timeout=35 + ) + found = { + name: revision + for line in output.splitlines() + for revision, name in [line.split()] + if name in patterns + and re.fullmatch(r"(?:[a-f0-9]{40}|[a-f0-9]{64})", revision) + } + if ref == "HEAD" or ref.startswith("refs/"): + return found.get(ref) + branch, tag = ( + found.get(patterns[0]), + found.get(patterns[2]) or found.get(patterns[1]), + ) + if branch and tag: + raise ValueError("Ambiguous branch and tag; use an explicit refs path") + return branch or tag + + return await asyncio.to_thread(read) + + +async def resolve_shared_source(uri, cache, *, refresh=False): + root = os.environ.get("AMPLIFIER_SOURCE_STORE") + if not root: + raise ValueError("No shared source store configured") + return await SharedSourceStore(root).resolve(uri, cache, refresh=refresh) + + +def build_view(source): + """A stable writable build input; immutable checkout remains untouched. + + Kept outside the object store, with exact Git provenance for qualification. + uv's wheel cache reuses artifacts; no editable package is shared by hosts. + """ + original = Path(source) + source = original.resolve() + for root in (source, *source.parents): + marker = root / ".amplifier_cache_meta.json" + if not marker.is_file() or not (root / ".git").is_dir(): + continue + data = json.loads(marker.read_text()) + if not data.get("immutable"): + return original, False + store = root.parent.parent.parent + target = store / "builds" / source_id(data["git_url"], data["commit"]) + from filelock import FileLock + + target.parent.mkdir(parents=True, exist_ok=True) + with FileLock(str(target) + ".lock"): + if not target.exists(): + shutil.copytree(root, target, symlinks=True) + for path in (target, *target.rglob("*")): + if not path.is_symlink(): + path.chmod(stat.S_IMODE(path.stat().st_mode) | 0o200) + atomic( + target / ".amplifier_cache_meta.json", + {**data, "immutable": False, "buildInput": True}, + ) + return target / source.relative_to(root), True + return original, False + + +def build_lock(source): + """A build backend may write metadata; serialize all users of its view.""" + from contextlib import nullcontext + + from filelock import FileLock + + source = Path(source).resolve() + for root in (source, *source.parents): + marker = root / ".amplifier_cache_meta.json" + if marker.is_file() and (root / ".git").is_dir(): + data = json.loads(marker.read_text()) + return ( + FileLock(str(root) + ".lock") + if data.get("buildInput") + else nullcontext() + ) + return nullcontext() diff --git a/docs/shared-source-store.md b/docs/shared-source-store.md new file mode 100644 index 00000000..146e1508 --- /dev/null +++ b/docs/shared-source-store.md @@ -0,0 +1,36 @@ +# Shared source store + +`AMPLIFIER_SOURCE_STORE` opts an application into commit-addressed Git source +storage. Without it, the existing resolver and installer contracts are unchanged. +The application owns the cache scope and its repository/ref bindings; Foundation +owns read-only exact-commit objects, per-object locks and atomic publication. + +`SharedSourceStore.bind(cache, url, ref, revision)` publishes a verified object +and writes a private binding under the application cache. Bundles and skills +share that binding scope. Ordinary resolution reads the binding; explicit update +resolves a new exact commit and replaces the binding without mutating old trees. +Branch/tag ambiguity requires an explicit ref. Credential-bearing URLs are not +adopted into shared storage. +The Git status API inspects the binding without cloning or replacing it. + +A legacy source is preserved until the application adopts a verified staging +copy. Dirty or ignored/untracked work and external Git directories cannot enter +the store. A dirty legacy checkout remains authoritative over a binding; refresh +fails without deleting the edit. The application must preflight every duplicate +of a repository/ref before adoption to avoid eclipsing a user's alternate copy. + +Module installation uses a writable build view with the same exact source +provenance and a process lock covering the build command. It installs wheels, +not editable references to the read-only tree. uv owns artifact caching and +resolution. Existing mutable/local sources preserve their editable behavior and +original install path. Stored objects and build views are retained: consumers +must prove all readers and references before implementing object collection. + +`AMPLIFIER_INSTALL_PREPARATION` can name one isolated preparation attempt with +a fresh 32-character hexadecimal token. With explicit dependency refresh, a +duplicate installation can reuse that attempt's receipt only when source content, +explicit constraints/overrides and the complete installed package metadata match. +An intervening graph/source/policy change or explicit force triggers installation. +A new attempt always resolves fresh; receipts do not pin future dependencies. +External install interpreters and unqualifiable external build inputs never reuse +the caller's metadata evidence. diff --git a/tests/test_dependency_refresh.py b/tests/test_dependency_refresh.py index 1f27ab7a..3b77e198 100644 --- a/tests/test_dependency_refresh.py +++ b/tests/test_dependency_refresh.py @@ -139,3 +139,68 @@ async def test_refresh_does_not_enable_disabled_installs(tmp_path, monkeypatch): install_deps=False, refresh_dependencies=True, cache_dir=tmp_path / "cache" ) install.assert_not_called() + + +@pytest.mark.asyncio +async def test_preparation_attempt_reuses_only_matching_source_policy_and_installed_graph( + tmp_path, monkeypatch +): + import sys + from amplifier_foundation.modules import preparation + + module = project(tmp_path / "module") + monkeypatch.setenv("AMPLIFIER_INSTALL_PREPARATION", "a" * 32) + graph = ["first"] + monkeypatch.setattr(preparation, "graph_signature", lambda: graph[0]) + activator = ModuleActivator( + cache_dir=tmp_path / "cache", + install_python=sys.executable, + refresh_dependencies=True, + ) + from unittest.mock import AsyncMock + + install = AsyncMock() + monkeypatch.setattr(activator, "_install_dependencies_uncached", install) + await activator._install_dependencies(module) + await activator._install_dependencies(module) + assert install.await_count == 1 + # Installing another module can change transitive dependencies. Recheck. + graph[0] = "changed graph" + await activator._install_dependencies(module) + assert install.await_count == 2 + (module / "implementation.py").write_text("source changed without a version bump") + await activator._install_dependencies(module) + assert install.await_count == 3 + # A fresh attempt never inherits earlier refresh evidence. + monkeypatch.setenv("AMPLIFIER_INSTALL_PREPARATION", "b" * 32) + await activator._install_dependencies(module) + assert install.await_count == 4 + await activator._install_dependencies(module, force=True) + assert install.await_count == 5 + + +@pytest.mark.asyncio +async def test_external_build_input_never_reuses_preparation_receipt( + tmp_path, monkeypatch +): + import sys + from unittest.mock import AsyncMock + from amplifier_foundation.modules import preparation + + module = project(tmp_path / "module") + external = tmp_path / "external.txt" + external.write_text("first") + (module / "external.txt").symlink_to(external) + monkeypatch.setenv("AMPLIFIER_INSTALL_PREPARATION", "a" * 32) + monkeypatch.setattr(preparation, "graph_signature", lambda: "same") + activator = ModuleActivator( + cache_dir=tmp_path / "cache", + install_python=sys.executable, + refresh_dependencies=True, + ) + install = AsyncMock() + monkeypatch.setattr(activator, "_install_dependencies_uncached", install) + await activator._install_dependencies(module) + external.write_text("changed") + await activator._install_dependencies(module) + assert install.await_count == 2 diff --git a/tests/test_dependency_refresh_integration.py b/tests/test_dependency_refresh_integration.py index 31618e06..aca8c0de 100644 --- a/tests/test_dependency_refresh_integration.py +++ b/tests/test_dependency_refresh_integration.py @@ -110,6 +110,54 @@ async def main(): ) +def test_shared_source_builds_real_wheel_without_editable_canonical_files(tmp_path): + import asyncio + from amplifier_foundation.sources.shared import SharedSourceStore + + env = { + name: os.environ[name] for name in ("PATH", "SystemRoot") if name in os.environ + } + env.update( + UV_NO_INDEX="true", + UV_NO_CONFIG="true", + UV_PYTHON_DOWNLOADS="never", + UV_CACHE_DIR=str(tmp_path / "uv-cache"), + PYTHONPATH=os.pathsep.join(path for path in sys.path if path), + ) + repo = repository(tmp_path / "source", "shared-fixture") + revision = run("git", "rev-parse", "HEAD", cwd=repo) + store = SharedSourceStore(tmp_path / "store") + cache = tmp_path / "generation/cache" + url = "https://example.invalid/shared-fixture" + canonical = asyncio.run(store.bind(cache, url, "main", revision, existing=repo)) + env["AMPLIFIER_SOURCE_STORE"] = str(store.root) + target = tmp_path / "worker" + run(UV, "venv", "--python", sys.executable, target, env=env) + override = tmp_path / "qualified.txt" + override.write_text("") + activate( + environment_python(target), + "git+" + url + "@main", + cache, + override, + env, + refresh=True, + ) + result = json.loads( + run( + environment_python(target), + "-c", + 'import json,importlib.metadata as m,shared_fixture; d=m.distribution("shared-fixture"); print(json.dumps({"value":shared_fixture.VALUE,"direct":json.loads(d.read_text("direct_url.json"))}))', + env=env, + ) + ) + assert result["value"] == "old" + assert not result["direct"].get("dir_info", {}).get("editable", False) + assert result["direct"]["url"].startswith((store.root / "builds").as_uri() + "/") + assert store.verify(url, revision) == canonical + assert not (canonical / "shared_fixture/__init__.py").stat().st_mode & 0o222 + + def test_new_generation_refreshes_same_version_direct_and_transitive_sources(tmp_path): # No indexes, network, external build dependencies, shared uv cache or host installs. # uv rejects --refresh together with --offline. Restrict resolution to our diff --git a/tests/test_shared_sources.py b/tests/test_shared_sources.py new file mode 100644 index 00000000..a25a1d0f --- /dev/null +++ b/tests/test_shared_sources.py @@ -0,0 +1,216 @@ +import json +import subprocess + +import pytest + +from amplifier_foundation.sources.shared import ( + SharedSourceStore, + build_view, + resolve_shared_source, +) + + +def git(root, *args): + return subprocess.check_output(["git", *args], cwd=root, text=True).strip() + + +@pytest.fixture +def repository(tmp_path): + repo = tmp_path / "repository" + repo.mkdir() + git(repo, "init", "-b", "main") + git(repo, "config", "user.name", "Fixture") + git(repo, "config", "user.email", "fixture@example.invalid") + (repo / "bundle.md").write_text("bundle") + (repo / "skills").mkdir() + (repo / "skills/SKILL.md").write_text("skill") + git(repo, "add", ".") + git(repo, "commit", "-m", "fixture") + return repo, git(repo, "rev-parse", "HEAD") + + +@pytest.mark.asyncio +async def test_bundle_and_skill_share_exact_snapshot_and_generation_bindings( + tmp_path, monkeypatch, repository +): + repo, first = repository + store = SharedSourceStore(tmp_path / "store") + monkeypatch.setenv("AMPLIFIER_SOURCE_STORE", str(store.root)) + url = "https://example.invalid/repo" + await store.bind( + tmp_path / "generation-one/cache", url, "main", first, existing=repo + ) + bundle = await resolve_shared_source( + "git+" + url + "@main", tmp_path / "generation-one/cache" + ) + skill = await resolve_shared_source( + "git+" + url + "@main#subdirectory=skills", + tmp_path / "generation-one/cache/skills", + ) + assert bundle.source_root == skill.source_root + assert skill.active_path == bundle.source_root / "skills" + (repo / "bundle.md").write_text("new bundle") + git(repo, "commit", "-am", "next") + second = git(repo, "rev-parse", "HEAD") + await store.bind( + tmp_path / "generation-two/cache", url, "main", second, existing=repo + ) + newer = await resolve_shared_source( + "git+" + url + "@main", tmp_path / "generation-two/cache" + ) + assert newer.source_root != bundle.source_root + assert (bundle.source_root / "bundle.md").read_text() == "bundle" + assert (newer.source_root / "bundle.md").read_text() == "new bundle" + assert not (bundle.source_root / "bundle.md").stat().st_mode & 0o222 + build, immutable = build_view(bundle.source_root) + assert immutable + (build / "bundle.md").write_text("generated build data") + assert (bundle.source_root / "bundle.md").read_text() == "bundle" + assert not json.loads((build / ".amplifier_cache_meta.json").read_text())[ + "immutable" + ] + + +@pytest.mark.asyncio +async def test_dirty_sources_and_credentials_do_not_enter_store(tmp_path, repository): + repo, revision = repository + store = SharedSourceStore(tmp_path / "store") + (repo / "bundle.md").write_text("local edit") + with pytest.raises(ValueError, match="clean"): + await store.ensure("https://example.invalid/repo", revision, existing=repo) + with pytest.raises(ValueError, match="credential-free"): + store.checkout("https://token@example.invalid/repo", revision) + assert not list((store.root / "objects").rglob(revision)) + + +@pytest.mark.asyncio +async def test_shared_snapshot_rejects_missing_subpath( + tmp_path, monkeypatch, repository +): + repo, revision = repository + store = SharedSourceStore(tmp_path / "store") + monkeypatch.setenv("AMPLIFIER_SOURCE_STORE", str(store.root)) + await store.bind( + tmp_path / "cache", + "https://example.invalid/repo", + "main", + revision, + existing=repo, + ) + with pytest.raises(ValueError, match="subpath"): + await resolve_shared_source( + "git+https://example.invalid/repo@main#subdirectory=missing", + tmp_path / "cache", + ) + + +@pytest.mark.asyncio +async def test_ignored_local_files_and_linked_worktrees_are_preserved( + tmp_path, repository +): + repo, revision = repository + (repo / ".git/info/exclude").write_text("local-notes.txt\n") + (repo / "local-notes.txt").write_text("retain private work") + store = SharedSourceStore(tmp_path / "store") + with pytest.raises(ValueError, match="clean"): + await store.ensure("https://example.invalid/repo", revision, existing=repo) + assert (repo / "local-notes.txt").read_text() == "retain private work" + + +@pytest.mark.asyncio +async def test_equivalent_urls_share_snapshot_and_refresh_never_mutates_reader( + tmp_path, monkeypatch, repository +): + repo, first = repository + store = SharedSourceStore(tmp_path / "store") + monkeypatch.setenv("AMPLIFIER_SOURCE_STORE", str(store.root)) + url = "https://example.invalid/repo" + old = await store.bind(tmp_path / "cache", url, "main", first, existing=repo) + assert await store.ensure(url + ".git", first) == old + (repo / "bundle.md").write_text("new bundle") + git(repo, "commit", "-am", "new") + second = git(repo, "rev-parse", "HEAD") + await store.ensure(url, second, existing=repo) + from amplifier_foundation.sources import shared + + async def remote(url, ref): + return second + + monkeypatch.setattr(shared, "remote_revision", remote) + result = await store.resolve( + "git+" + url + "@main", tmp_path / "cache", refresh=True + ) + assert result.source_root != old + assert (old / "bundle.md").read_text() == "bundle" + assert (result.source_root / "bundle.md").read_text() == "new bundle" + + +def test_object_symlink_cannot_redirect_shared_storage(tmp_path, repository): + repo, revision = repository + store = SharedSourceStore(tmp_path / "store") + target = store.checkout("https://example.invalid/repo", revision) + target.parent.mkdir(parents=True) + target.symlink_to(repo, target_is_directory=True) + with pytest.raises(ValueError, match="owned store"): + store.verify("https://example.invalid/repo", revision) + assert (repo / "bundle.md").read_text() == "bundle" + + +@pytest.mark.asyncio +async def test_nested_legacy_bundle_edits_remain_authoritative(tmp_path, repository): + import shutil + + from amplifier_foundation.paths.resolution import parse_uri + from amplifier_foundation.sources.git import GitSourceHandler + + repo, revision = repository + store = SharedSourceStore(tmp_path / "store") + cache = tmp_path / "generation/cache/bundles" + uri = "git+https://example.invalid/repo@main" + await store.bind( + cache, "https://example.invalid/repo", "main", revision, existing=repo + ) + legacy = GitSourceHandler()._get_cache_path(parse_uri(uri), cache) + shutil.copytree(repo, legacy) + (legacy / "bundle.md").write_text("local bundle edit") + result = await store.resolve(uri, cache) + assert result.source_root == legacy + assert (result.active_path / "bundle.md").read_text() == "local bundle edit" + with pytest.raises(ValueError, match="Local changes"): + await store.resolve(uri, cache, refresh=True) + + +@pytest.mark.asyncio +async def test_shared_status_reads_binding_without_clone_or_mutation( + tmp_path, monkeypatch, repository +): + from amplifier_foundation.paths.resolution import parse_uri + from amplifier_foundation.sources import shared + from amplifier_foundation.sources.git import GitSourceHandler + + repo, revision = repository + store = SharedSourceStore(tmp_path / "store") + cache = tmp_path / "generation/cache/bundles" + await store.bind( + cache, "https://example.invalid/repo", "main", revision, existing=repo + ) + monkeypatch.setenv("AMPLIFIER_SOURCE_STORE", str(store.root)) + + async def remote(url, ref): + assert ref == "main" + return revision + + monkeypatch.setattr(shared, "remote_revision", remote) + before = list(store.root.rglob("*")) + # Equivalent .git spellings retain the same binding and status contract. + uri = "git+https://example.invalid/repo.git@main" + status = await GitSourceHandler().get_status(parse_uri(uri), cache) + assert ( + status.is_cached + and status.cached_commit == revision + and status.has_update is False + ) + assert list(store.root.rglob("*")) == before + assert (await store.resolve(uri, cache)).source_root == store.checkout( + "https://example.invalid/repo", revision + ) From 7582f0831a15f67dde3676c8765d3e882e4fd881 Mon Sep 17 00:00:00 2001 From: Brian Krabach Date: Thu, 1 Oct 2026 05:38:51 -0700 Subject: [PATCH 2/3] Support long shared object paths and isolated homes on Windows --- amplifier_foundation/sources/shared.py | 36 ++++++++++++++------ tests/test_dependency_refresh_integration.py | 3 ++ 2 files changed, 28 insertions(+), 11 deletions(-) diff --git a/amplifier_foundation/sources/shared.py b/amplifier_foundation/sources/shared.py index 32ca37a0..c0595e8c 100644 --- a/amplifier_foundation/sources/shared.py +++ b/amplifier_foundation/sources/shared.py @@ -22,6 +22,7 @@ from filelock import AsyncFileLock from amplifier_foundation.paths.resolution import ResolvedSource, parse_uri +from amplifier_foundation.sources.git import _with_longpaths async def complete_io(function, *args, **kwargs): @@ -59,13 +60,22 @@ def clean_checkout(root): root = Path(root) if not (root / ".git").is_dir() or (root / ".git").is_symlink(): return False # Linked worktrees can refer to a mutable external Git dir. + # Commit-addressed roots can exceed MAX_PATH even for Git object reads. dirty = subprocess.check_output( - ["git", "--no-optional-locks", "status", "--porcelain", "--untracked-files=no"], + _with_longpaths( + [ + "git", + "--no-optional-locks", + "status", + "--porcelain", + "--untracked-files=no", + ] + ), cwd=root, text=True, ) extras = subprocess.check_output( - ["git", "ls-files", "--others", "-z"], cwd=root + _with_longpaths(["git", "ls-files", "--others", "-z"]), cwd=root ).split(b"\0") extras = [name for name in extras if name and name != b".amplifier_cache_meta.json"] return not dirty and not extras @@ -92,7 +102,7 @@ def verify(self, url, revision): ): raise ValueError("Shared source identity changed") actual = subprocess.check_output( - ["git", "rev-parse", "HEAD"], cwd=target, text=True + _with_longpaths(["git", "rev-parse", "HEAD"]), cwd=target, text=True ).strip() if actual != revision or not clean_checkout(target): raise ValueError("Shared source contents changed") @@ -136,16 +146,20 @@ def publish(): }, ) actual = subprocess.check_output( - ["git", "rev-parse", "HEAD"], cwd=stage, text=True + _with_longpaths(["git", "rev-parse", "HEAD"]), + cwd=stage, + text=True, ).strip() dirty = subprocess.check_output( - [ - "git", - "--no-optional-locks", - "status", - "--porcelain", - "--untracked-files=no", - ], + _with_longpaths( + [ + "git", + "--no-optional-locks", + "status", + "--porcelain", + "--untracked-files=no", + ] + ), cwd=stage, text=True, ) diff --git a/tests/test_dependency_refresh_integration.py b/tests/test_dependency_refresh_integration.py index aca8c0de..2a1b86f4 100644 --- a/tests/test_dependency_refresh_integration.py +++ b/tests/test_dependency_refresh_integration.py @@ -117,6 +117,9 @@ def test_shared_source_builds_real_wheel_without_editable_canonical_files(tmp_pa env = { name: os.environ[name] for name in ("PATH", "SystemRoot") if name in os.environ } + fixture_home = tmp_path / "home" + fixture_home.mkdir() + env.update(HOME=str(fixture_home), USERPROFILE=str(fixture_home)) env.update( UV_NO_INDEX="true", UV_NO_CONFIG="true", From 5ef75e207ea858d220e562790e658e469a57b014 Mon Sep 17 00:00:00 2001 From: Brian Krabach Date: Thu, 1 Oct 2026 06:54:05 -0700 Subject: [PATCH 3/3] Add independent registry read policy --- amplifier_foundation/registry.py | 10 ++++++++-- docs/shared-source-store.md | 6 ++++++ tests/test_registry.py | 29 +++++++++++++++++++++++++++++ 3 files changed, 43 insertions(+), 2 deletions(-) diff --git a/amplifier_foundation/registry.py b/amplifier_foundation/registry.py index 43623a58..ba6ed6ea 100644 --- a/amplifier_foundation/registry.py +++ b/amplifier_foundation/registry.py @@ -185,6 +185,7 @@ def __init__( strict: bool = False, include_source_resolver: Callable[[str], str | None] | None = None, persist: bool = True, + read_persisted: bool = True, ) -> None: """Initialize registry. @@ -204,6 +205,10 @@ def __init__( cleanup and explicit save() calls cannot write registry.json. Source downloads still use the shared content cache. Use a fresh instance per session with scoped source overrides. + read_persisted: If False, start with no saved registrations. The + caller supplies its authoritative registrations; the source + cache still belongs to the explicit home. Defaults to True + so existing CLI and library consumers retain their behavior. """ self._home = self._resolve_home(home) self._strict = strict @@ -217,8 +222,9 @@ def __init__( # Future-based deduplication: cache loaded bundles and track in-progress loads self._loaded_bundles: dict[str, Bundle] = {} # Cache of fully loaded bundles self._pending_loads: dict[str, asyncio.Future[Bundle]] = {} # In-progress loads - self._load_persisted_state() - self._validate_cached_paths() + if read_persisted: + self._load_persisted_state() + self._validate_cached_paths() @property def home(self) -> Path: diff --git a/docs/shared-source-store.md b/docs/shared-source-store.md index 146e1508..5a4d410c 100644 --- a/docs/shared-source-store.md +++ b/docs/shared-source-store.md @@ -34,3 +34,9 @@ An intervening graph/source/policy change or explicit force triggers installatio A new attempt always resolves fresh; receipts do not pin future dependencies. External install interpreters and unqualifiable external build inputs never reuse the caller's metadata evidence. + +Hosts that construct registrations from shared settings can create +`BundleRegistry(home=owned_home, persist=False, read_persisted=False)`. +This skips saved registrations even when an older imported registry remains in +that directory. Both flags default to `True`, preserving existing CLI/library +behavior. The explicit `home` still owns source caches. diff --git a/tests/test_registry.py b/tests/test_registry.py index b131ab12..8bde0b45 100644 --- a/tests/test_registry.py +++ b/tests/test_registry.py @@ -1977,3 +1977,32 @@ async def waiter() -> None: ) assert isinstance(waiter_exc, RuntimeError) assert sentinel in str(waiter_exc) + + +def test_independent_registry_skips_saved_aliases_without_changing_default(tmp_path): + """Hosts can use settings authority without importing another registry.""" + import json + + path = tmp_path / "registry.json" + path.write_text(json.dumps({"version": 1, "bundles": { + "legacy": {"uri": "git+https://example.invalid/legacy@main"}}})) + before = path.read_bytes() + default = BundleRegistry(home=tmp_path, persist=False) + assert default.find("legacy") == "git+https://example.invalid/legacy@main" + independent = BundleRegistry(home=tmp_path, persist=False, read_persisted=False) + assert independent.list_registered() == [] + independent.register({"selected": "git+https://example.invalid/selected@main"}) + assert independent.find("selected") == "git+https://example.invalid/selected@main" + independent.save() + assert path.read_bytes() == before + + +def test_independent_registry_never_reads_saved_state(tmp_path, monkeypatch): + from amplifier_foundation import BundleRegistry + + def forbidden(*args): + raise AssertionError("Saved registration state must not be inspected") + + monkeypatch.setattr(BundleRegistry, "_load_persisted_state", forbidden) + monkeypatch.setattr(BundleRegistry, "_validate_cached_paths", forbidden) + assert BundleRegistry(home=tmp_path, read_persisted=False).list_registered() == []