feat(cache): restore and publish the cargo dependency cache around Linux builds

Restore runs after the source is in place and before the build; publish runs
only after a successful build and only on a miss, so the steady state uploads
nothing and a failed build can never become an entry. The whole path is
best-effort: a cache problem degrades to a cold build with a message rather
than turning a good build red.

Gated on Cargo.lock being present, so non-Rust jobs are untouched with no flag
to set. The Windows branch is not touched at all.

The prune script is read through importlib.resources rather than a path
relative to the repo root: the production venv is installed non-editable, so a
parents[] walk lands in site-packages where scripts/ does not exist.

The fake transport gained substring-matched canned results; indexing by call
position broke as soon as the cache added calls mid-flow.

Mutation-tested rather than assumed. Three guards were each disabled in turn --
the publish call, the prune-refusal check, the packing-failure check -- and a
test went red for each. The prune-refusal test initially survived its mutant:
it passed because the fake's fetch left no file behind, so publishing failed
for an unrelated reason. Fixed by making fetch write a real file, so the only
thing preventing publication is the guard under test.

src/ci_orchestrator/commands/build.py

@@ -25,11 +25,13 @@ import tarfile
 import tempfile
 import zipfile
 from collections.abc import Iterable
+from importlib import resources
 from pathlib import Path
 from typing import TYPE_CHECKING
 
 import click
 
+from ci_orchestrator.cargo_cache import CargoCacheStore, cache_key
 from ci_orchestrator.config import load_config
 from ci_orchestrator.credentials import KeyringCredentialStore
 from ci_orchestrator.transport.errors import TransportCommandError, TransportError
@@ -122,6 +124,110 @@ def _tail_lines(stdout: str, stderr: str) -> str:
     return " | ".join(lines[-5:]) if lines else "(no build output)"
 
 
+# ------------------------------------------------- cargo dependency cache
+
+_PRUNE_ASSET = "cargo-cache-prune.sh"
+_GUEST_PRUNE = "/tmp/cargo-cache-prune.sh"
+_GUEST_TARBALL = "/tmp/cargo-cache-entry.tar"
+
+
+def _cargo_cache_probe(t: SshTransport, workdir: str) -> tuple[str, str] | None:
+    """Return ``(key, compressor)`` for this workspace, or ``None`` to skip.
+
+    ``None`` means "not a Rust build, or we could not read what we need". The
+    cache is an optimisation, so every uncertainty here resolves to skipping
+    it rather than to guessing.
+    """
+    if t.run(f"test -f {_sh_quote(workdir + '/Cargo.lock')}", check=False).returncode != 0:
+        return None
+    lock = t.run(f"sha256sum {_sh_quote(workdir + '/Cargo.lock')}", check=False)
+    version = t.run("rustc -V", check=False)
+    triple = t.run("rustc -vV | sed -n 's/^host: //p'", check=False)
+    if not (lock.ok and version.ok and triple.ok):
+        click.echo("[cache] could not read Cargo.lock digest or rustc version; skipping")
+        return None
+    digest = lock.stdout.split()[0] if lock.stdout.split() else ""
+    if not digest:
+        click.echo("[cache] empty Cargo.lock digest; skipping")
+        return None
+    compressor = "zstd" if t.run("command -v zstd", check=False).ok else "gzip"
+    key = cache_key(
+        lock_digest=digest,
+        rustc_version=version.stdout.strip(),
+        host_triple=triple.stdout.strip(),
+    )
+    return key, compressor
+
+
+def _tar_flag(compressor: str) -> tuple[str, str]:
+    """``(tar flag, file extension)`` for the chosen compressor."""
+    return ("--zstd", "zst") if compressor == "zstd" else ("-z", "gz")
+
+
+def _cargo_cache_restore(
+    t: SshTransport, store: CargoCacheStore, key: str, workdir: str, compressor: str
+) -> bool:
+    """Upload and unpack an existing entry. Returns True on a hit."""
+    entry = store.lookup(key)
+    if entry is None:
+        click.echo(f"[cache] miss for {key} — building cold")
+        return False
+    flag, ext = _tar_flag(compressor)
+    size_mb = entry.stat().st_size / 1_048_576
+    click.echo(f"[cache] hit for {key} ({size_mb:.0f} MB) — restoring")
+    guest_entry = f"{_GUEST_TARBALL}.{ext}"
+    t.copy(str(entry), guest_entry)
+    t.run(
+        f"tar {flag} -xf {_sh_quote(guest_entry)} -C {_sh_quote(workdir)} && "
+        f"rm -f {_sh_quote(guest_entry)}"
+    )
+    return True
+
+
+def _cargo_cache_publish(
+    t: SshTransport, store: CargoCacheStore, key: str, workdir: str, compressor: str
+) -> None:
+    """Prune workspace artifacts, then file the remainder under ``key``.
+
+    A non-zero prune means the fail-closed assertion tripped: say so loudly and
+    publish nothing. A poisoned entry costs a wrong verdict; a missing one only
+    costs a cold build.
+    """
+    # as_file() because the production venv is installed non-editable: the
+    # script is package data, not a path relative to a source checkout.
+    asset = resources.files("ci_orchestrator.assets").joinpath(_PRUNE_ASSET)
+    with resources.as_file(asset) as prune_script:
+        t.copy(str(prune_script), _GUEST_PRUNE)
+    pruned = t.run(
+        f"bash {_GUEST_PRUNE} {_sh_quote(workdir + '/target')} "
+        f"{_sh_quote(workdir + '/Cargo.toml')}",
+        check=False,
+    )
+    click.echo(f"[cache] {(pruned.stdout.strip() or pruned.stderr.strip())}")
+    if pruned.returncode != 0:
+        click.echo("[cache] prune refused — publishing nothing", err=True)
+        return
+
+    flag, ext = _tar_flag(compressor)
+    guest_tar = f"{_GUEST_TARBALL}.{ext}"
+    packed = t.run(
+        f"tar {flag} -cf {_sh_quote(guest_tar)} -C {_sh_quote(workdir)} target",
+        check=False,
+    )
+    if not packed.ok:
+        click.echo("[cache] packing failed — publishing nothing", err=True)
+        return
+
+    with tempfile.TemporaryDirectory() as td:
+        local = Path(td) / f"entry.tar.{ext}"
+        t.fetch(guest_tar, local)
+        published = store.publish(key, local)
+    t.run(f"rm -f {_sh_quote(guest_tar)}", check=False)
+    click.echo(
+        f"[cache] published {key} ({published.stat().st_size / 1_048_576:.0f} MB)"
+    )
+
+
 # ---------------------------------------------------------------- Linux flow
 
 
@@ -143,6 +249,7 @@ def _linux_build(
     extra_env: dict[str, str],
     skip_artifact: bool,
     use_xvfb: bool = False,
+    cargo_cache_root: Path | None = None,
 ) -> None:
     click.echo("[build run] Linux build mode (SSH transport)")
 
@@ -217,6 +324,26 @@ def _linux_build(
             check=False,
         )
 
+        # Cargo dependency cache. Everything here is best-effort: a cache
+        # problem must never turn a good build red, so any failure degrades to
+        # a cold build with a message.
+        cache_key_value: str | None = None
+        compressor = "gzip"
+        store: CargoCacheStore | None = None
+        if cargo_cache_root is not None:
+            try:
+                store = CargoCacheStore(cargo_cache_root)
+                probed = _cargo_cache_probe(t, workdir)
+                if probed is None:
+                    store = None
+                else:
+                    cache_key_value, compressor = probed
+                    _cargo_cache_restore(t, store, cache_key_value, workdir, compressor)
+            except (TransportError, OSError, ValueError) as exc:
+                click.echo(f"[cache] restore skipped: {exc}", err=True)
+                cache_key_value = None
+                store = None
+
         env_prefix = "".join(
             f"export {k}={_sh_quote(v)}; " for k, v in extra_env.items()
         )
@@ -245,6 +372,14 @@ def _linux_build(
                 f"(exit {result.returncode}): {_tail_lines(result.stdout, result.stderr)}"
             )
 
+        # Publish only on a miss, so the steady state uploads nothing. A failed
+        # build never reaches here, so a broken tree cannot become an entry.
+        if store is not None and cache_key_value and store.lookup(cache_key_value) is None:
+            try:
+                _cargo_cache_publish(t, store, cache_key_value, workdir, compressor)
+            except (TransportError, OSError, ValueError) as exc:
+                click.echo(f"[cache] publish skipped: {exc}", err=True)
+
         if skip_artifact:
             click.echo("[build run] artifact packaging skipped (--skip-artifact).")
             return

tests/python/test_commands_build.py

@@ -38,9 +38,20 @@ class _FakeTransport:
         self.copies: list[tuple[str, str]] = []
         self.fetches: list[tuple[str, str]] = []
         self.run_results: dict[int, _FakeResult] = {}
+        # Matched by substring against the script, checked before the
+        # positional run_results. Index-based fixtures get brittle once a
+        # feature adds calls in the middle of the flow (the cargo cache does).
+        self.script_results: list[tuple[str, _FakeResult]] = []
         self.default_result = _FakeResult(stdout="ok")
         type(self).instances.append(self)
 
+    def result_for(
+        self, needle: str, *, stdout: str = "", stderr: str = "", returncode: int = 0
+    ) -> None:
+        self.script_results.append(
+            (needle, _FakeResult(stdout=stdout, stderr=stderr, returncode=returncode))
+        )
+
     def __enter__(self) -> _FakeTransport:
         return self
 
@@ -49,6 +60,12 @@ class _FakeTransport:
 
     def run(self, script: str, *, check: bool = True) -> _FakeResult:
         self.runs.append((script, check))
+        result: _FakeResult | None = None
+        for needle, canned in self.script_results:
+            if needle in script:
+                result = canned
+                break
+        if result is None:
             result = self.run_results.get(len(self.runs) - 1, self.default_result)
         if check and not result.ok:
             raise TransportCommandError(result.returncode, result.stdout, result.stderr)
@@ -838,3 +855,205 @@ def test_build_run_windows_transport_error(
     )
     assert result.exit_code != 0
     assert "winrm unreachable" in result.output
+
+
+# ── _linux_build: cargo dependency cache ───────────────────────────────────
+
+
+def _linux_kwargs(**overrides: Any) -> dict[str, Any]:
+    base: dict[str, Any] = {
+        "ip_address": "10.0.0.1",
+        "ssh_user": "ci_build",
+        "key_path": None,
+        "known_hosts": None,
+        "workdir": "/opt/ci/build",
+        "output_dir": "/opt/ci/output",
+        "host_source_dir": None,
+        "clone_url": "http://repo/example.git",
+        "clone_branch": "main",
+        "clone_commit": "",
+        "clone_submodules": False,
+        "build_command": "true",
+        "artifact_source": "dist",
+        "extra_env": {},
+        "skip_artifact": True,
+    }
+    base.update(overrides)
+    return base
+
+
+def _rust_guest(fake: _FakeTransport, *, digest: str = "cafe") -> None:
+    """Make the fake answer like a Rust workspace with a known toolchain."""
+    fake.result_for("sha256sum", stdout=f"{digest}  Cargo.lock\n")
+    fake.result_for("rustc -V", stdout="rustc 1.97.1 (abc 2026-01-01)\n")
+    fake.result_for("rustc -vV", stdout="x86_64-unknown-linux-gnu\n")
+
+
+def _expected_key(digest: str = "cafe") -> str:
+    from ci_orchestrator.cargo_cache import cache_key
+
+    return cache_key(
+        lock_digest=digest,
+        rustc_version="rustc 1.97.1 (abc 2026-01-01)",
+        host_triple="x86_64-unknown-linux-gnu",
+    )
+
+
+def _fetch_writes_a_real_file(
+    monkeypatch: pytest.MonkeyPatch, fake: _FakeTransport
+) -> None:
+    """Make ``fetch`` produce an actual file on disk.
+
+    Without this the default fake leaves nothing behind, ``publish`` dies on
+    FileNotFoundError, and a "nothing was published" assertion passes for the
+    wrong reason -- it would hold even with the guard under test removed.
+    Mutation-tested: dropping the prune-refusal check must make those tests go
+    red, and it only does when a publish could otherwise have succeeded.
+    """
+
+    def _fetch(remote: str, local: str) -> None:
+        fake.fetches.append((remote, str(local)))
+        Path(local).write_bytes(b"fresh-entry")
+
+    monkeypatch.setattr(fake, "fetch", _fetch)
+
+
+def test_linux_build_without_cache_root_runs_no_cache_commands(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(build_module, "SshTransport", _FakeTransport)
+    build_module._linux_build(**_linux_kwargs())
+    cmds = [c[0] for c in _FakeTransport.instances[0].runs]
+    assert not any("cargo-cache-prune" in c for c in cmds)
+    assert not any("sha256sum" in c for c in cmds)
+
+
+def test_linux_build_skips_cache_when_no_cargo_lock(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    monkeypatch.setattr(build_module, "SshTransport", _FakeTransport)
+    fake = _FakeTransport()
+    _FakeTransport.instances = []
+    monkeypatch.setattr(build_module, "SshTransport", lambda *a, **k: fake)
+    fake.result_for("test -f", returncode=1)
+    build_module._linux_build(**_linux_kwargs(cargo_cache_root=tmp_path))
+    cmds = [c[0] for c in fake.runs]
+    assert not any("cargo-cache-prune" in c for c in cmds)
+    assert not any("sha256sum" in c for c in cmds)
+
+
+def test_linux_build_restores_an_existing_entry(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    from ci_orchestrator.cargo_cache import CargoCacheStore
+
+    fake = _FakeTransport()
+    monkeypatch.setattr(build_module, "SshTransport", lambda *a, **k: fake)
+    _rust_guest(fake)
+
+    store = CargoCacheStore(tmp_path)
+    seed = tmp_path / "seed.tar.zst"
+    seed.write_bytes(b"entry")
+    store.publish(_expected_key(), seed)
+
+    build_module._linux_build(**_linux_kwargs(cargo_cache_root=tmp_path))
+    cmds = [c[0] for c in fake.runs]
+    assert any("tar --zstd -xf" in c for c in cmds), "a hit must unpack the entry"
+    assert fake.copies, "a hit must upload the entry to the guest"
+    # A hit must not re-publish: the entry is already there.
+    assert not any("cargo-cache-prune" in c for c in cmds)
+
+
+def test_linux_build_publishes_on_a_miss(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    from ci_orchestrator.cargo_cache import CargoCacheStore
+
+    fake = _FakeTransport()
+    monkeypatch.setattr(build_module, "SshTransport", lambda *a, **k: fake)
+    _rust_guest(fake)
+
+    _fetch_writes_a_real_file(monkeypatch, fake)
+
+    build_module._linux_build(**_linux_kwargs(cargo_cache_root=tmp_path))
+    assert CargoCacheStore(tmp_path).lookup(_expected_key()) is not None
+
+
+def test_linux_build_does_not_publish_when_the_prune_assertion_fails(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    from ci_orchestrator.cargo_cache import CargoCacheStore
+
+    fake = _FakeTransport()
+    monkeypatch.setattr(build_module, "SshTransport", lambda *a, **k: fake)
+    _rust_guest(fake)
+    _fetch_writes_a_real_file(monkeypatch, fake)
+    fake.result_for(
+        "cargo-cache-prune",
+        returncode=2,
+        stderr="prune: REFUSING to publish -- artifacts still reference workspace source",
+    )
+
+    build_module._linux_build(**_linux_kwargs(cargo_cache_root=tmp_path))
+    assert CargoCacheStore(tmp_path).lookup(_expected_key()) is None, (
+        "a failed prune assertion must never reach the store"
+    )
+
+
+def test_linux_build_does_not_publish_when_packing_fails(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    from ci_orchestrator.cargo_cache import CargoCacheStore
+
+    fake = _FakeTransport()
+    monkeypatch.setattr(build_module, "SshTransport", lambda *a, **k: fake)
+    _rust_guest(fake)
+    _fetch_writes_a_real_file(monkeypatch, fake)
+    fake.result_for("tar --zstd -cf", returncode=1, stderr="no space left on device")
+
+    build_module._linux_build(**_linux_kwargs(cargo_cache_root=tmp_path))
+    assert CargoCacheStore(tmp_path).lookup(_expected_key()) is None
+
+
+def test_cache_probe_failure_does_not_fail_the_build(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    fake = _FakeTransport()
+    monkeypatch.setattr(build_module, "SshTransport", lambda *a, **k: fake)
+    fake.result_for("sha256sum", returncode=1, stderr="boom")
+    # Must not raise: the cache is an optimisation, never a failure mode.
+    build_module._linux_build(**_linux_kwargs(cargo_cache_root=tmp_path))
+
+
+def test_cache_transport_error_does_not_fail_the_build(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    fake = _FakeTransport()
+    monkeypatch.setattr(build_module, "SshTransport", lambda *a, **k: fake)
+    _rust_guest(fake)
+
+    def _boom(*_a: Any, **_k: Any) -> None:
+        raise TransportError("ssh died mid-restore")
+
+    monkeypatch.setattr(fake, "copy", _boom)
+    build_module._linux_build(**_linux_kwargs(cargo_cache_root=tmp_path))
+
+
+def test_cache_key_differs_when_the_lockfile_differs(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    """A Cargo.lock change must miss the entry built for the previous one."""
+    from ci_orchestrator.cargo_cache import CargoCacheStore
+
+    store = CargoCacheStore(tmp_path)
+    seed = tmp_path / "seed.tar.zst"
+    seed.write_bytes(b"entry")
+    store.publish(_expected_key("cafe"), seed)
+
+    fake = _FakeTransport()
+    monkeypatch.setattr(build_module, "SshTransport", lambda *a, **k: fake)
+    _rust_guest(fake, digest="beef")
+
+    build_module._linux_build(**_linux_kwargs(cargo_cache_root=tmp_path))
+    cmds = [c[0] for c in fake.runs]
+    assert not any("tar --zstd -xf" in c for c in cmds), "different lock must miss"