feat(agents): add incremental zstd driver updates

.github/workflows/agents-release.yml

@@ -1631,9 +1631,26 @@ jobs:
       - name: Verify managed JDBC assets in all offline ZIPs
         run: node agents/scripts/verify_offline_jdbc_release.mjs release
 
+      - name: Fetch previous agent registry for delta patches
+        env:
+          GH_TOKEN: ${{ steps.release-token.outputs.token }}
+        run: |
+          mkdir -p prev-release
+          if gh release download agents-latest --repo "$GITHUB_REPOSITORY" \
+              --pattern agent-registry.json --dir prev-release --clobber; then
+            echo "previous agent-registry.json fetched for delta generation"
+          else
+            echo "no previous agents-latest registry found; delta patches are skipped"
+            rm -rf prev-release
+          fi
+
       - name: Build single-driver packages
         run: |
-          python3 agents/scripts/build_driver_zips.py release --cleanup-sources
+          PREV_ARGS=()
+          if [ -f prev-release/agent-registry.json ]; then
+            PREV_ARGS+=(--previous-registry prev-release/agent-registry.json)
+          fi
+          python3 agents/scripts/build_driver_zips.py release --cleanup-sources "${PREV_ARGS[@]}"
           python3 -m json.tool release/agent-registry.json > /dev/null
           echo "=== final agent-registry.json ==="
           cat release/agent-registry.json
@@ -1830,6 +1847,7 @@ jobs:
             --recursive \
             --exclude "*" \
             --include "dbx-agent-*.tar.zst" \
+            --include "dbx-agent-*.tar.zst.delta" \
             --endpoint-url "$R2_ENDPOINT"
           echo "Uploaded single-driver packages"
 

agents/scripts/build_driver_zips.py

@@ -5,8 +5,10 @@ import hashlib
 import io
 import json
 import subprocess
+import sys
 import tarfile
 import tempfile
+import urllib.request
 from pathlib import Path
 from urllib.parse import urlparse
 
@@ -62,13 +64,148 @@ def update_release_artifact(artifact: dict, output: Path) -> None:
     artifact["format"] = "tar_zstd"
 
 
-def build_driver_zips(release_dir: Path) -> list[Path]:
+# Incremental updates only pay off for packages big enough to matter on slow
+# links; small packages ship a full download either way.
+MIN_DELTA_FULL_SIZE = 8 * 1024 * 1024
+# A delta larger than this fraction of the full package is not worth hosting:
+# clients would rather re-download the full artifact.
+MAX_DELTA_RATIO = 0.5
+
+
+class DeltaGenerator:
+    """Generates `zstd --patch-from` deltas between two releases of one artifact.
+
+    The previous registry (from the `agents-latest` release) provides each
+    driver's previously published version and artifact URL. For every artifact
+    whose version changed, the previous package is downloaded once, the new
+    package is compressed against it, and — when the delta is small enough to
+    be worth it — both the `.delta` file and the registry `delta` block are
+    emitted. Any failure to produce a delta is logged and skipped: the release
+    must never break because an incremental update could not be built.
+    """
+
+    def __init__(
+        self,
+        release_dir: Path,
+        previous_registry: dict | None,
+        *,
+        min_full_size: int = MIN_DELTA_FULL_SIZE,
+        max_ratio: float = MAX_DELTA_RATIO,
+        verbose: bool = True,
+    ) -> None:
+        self.release_dir = release_dir
+        self.previous_registry = previous_registry or {"drivers": {}}
+        self.min_full_size = min_full_size
+        self.max_ratio = max_ratio
+        self.verbose = verbose
+        self._downloads: dict[str, Path] = {}
+        self._download_dir: Path | None = None
+
+    def _previous_driver(self, driver_name: str) -> dict | None:
+        return self.previous_registry.get("drivers", {}).get(driver_name)
+
+    def _download_previous(self, url: str) -> Path | None:
+        if url in self._downloads:
+            return self._downloads[url]
+        if self._download_dir is None:
+            self._download_dir = Path(tempfile.mkdtemp(prefix="dbx-delta-prev-"))
+        target = self._download_dir / artifact_filename(url)
+        try:
+            with urllib.request.urlopen(url) as response, target.open("wb") as out:
+                out.write(response.read())
+        except Exception as error:  # noqa: BLE001 - any fetch failure just skips the delta
+            print(f"  delta skipped (previous artifact unavailable: {error})", file=sys.stderr)
+            return None
+        self._downloads[url] = target
+        return target
+
+    def maybe_generate(
+        self,
+        driver_name: str,
+        new_version: str,
+        new_artifact: dict,
+        previous_artifact: dict | None,
+        new_package: Path,
+        suffix: str,
+    ) -> None:
+        """Attaches a `delta` block to the freshly packaged `new_artifact`.
+
+        `previous_artifact` is the same artifact slot (jar, or one native
+        platform) in the previous registry and only supplies the base package
+        URL; `suffix` distinguishes native platforms in the delta file name.
+        """
+        if not previous_artifact:
+            return
+        if new_package.stat().st_size < self.min_full_size:
+            return
+        previous_driver = self._previous_driver(driver_name) or {}
+        previous_version = str(previous_driver.get("version", ""))
+        if not previous_version or previous_version == new_version:
+            return
+        base_url = previous_artifact.get("url", "")
+        if not base_url:
+            return
+        previous_package = self._download_previous(base_url)
+        if previous_package is None:
+            return
+
+        stem = f"dbx-agent-{driver_name}-{previous_version}-to-{new_version}{suffix}"
+        delta_path = self.release_dir / f"{stem}.tar.zst.delta"
+        try:
+            subprocess.run(
+                [
+                    "zstd", "-q", "--patch-from", str(previous_package),
+                    str(new_package), "-o", str(delta_path), "--force",
+                ],
+                check=True,
+            )
+        except (OSError, subprocess.CalledProcessError) as error:
+            print(f"  delta skipped (zstd failed: {error})", file=sys.stderr)
+            delta_path.unlink(missing_ok=True)
+            return
+
+        delta_size = delta_path.stat().st_size
+        if delta_size > self.max_ratio * new_package.stat().st_size:
+            delta_path.unlink()
+            if self.verbose:
+                ratio = delta_size / new_package.stat().st_size
+                print(f"  delta skipped ({ratio:.0%} of full package, above {self.max_ratio:.0%})")
+            return
+
+        # `new_artifact["url"]` already points at the packaged tar.zst on this
+        # release; the delta file rides the same release, so only the file name
+        # needs swapping.
+        new_artifact["delta"] = {
+            "base_version": previous_version,
+            "url": release_url_with_filename(new_artifact["url"], delta_path.name),
+            "sha256": hashlib.sha256(delta_path.read_bytes()).hexdigest(),
+            "size": delta_size,
+        }
+        if self.verbose:
+            ratio = delta_size / new_package.stat().st_size
+            print(f"  delta {delta_path.name} ({delta_size} bytes, {ratio:.0%} of full)")
+
+
+def build_driver_zips(
+    release_dir: Path,
+    previous_registry: dict | None = None,
+    *,
+    delta_min_full_size: int = MIN_DELTA_FULL_SIZE,
+    delta_max_ratio: float = MAX_DELTA_RATIO,
+) -> list[Path]:
     registry_path = release_dir / "agent-registry.json"
     registry = json.loads(registry_path.read_text(encoding="utf-8"))
+    delta_generator = DeltaGenerator(
+        release_dir,
+        previous_registry,
+        min_full_size=delta_min_full_size,
+        max_ratio=delta_max_ratio,
+    )
     outputs: list[Path] = []
 
     for driver_name, driver in registry.get("drivers", {}).items():
         version = driver["version"]
+        previous_driver = delta_generator._previous_driver(driver_name)
         jar_artifact = driver.get("jar")
         if jar_artifact and jar_artifact.get("size", 0) > 0:
             filename = artifact_filename(jar_artifact["url"])
@@ -86,6 +223,10 @@ def build_driver_zips(release_dir: Path) -> list[Path]:
             elif not output.is_file():
                 raise FileExistsError(f"Reusable Java agent package is not a file: {output}")
             update_release_artifact(jar_artifact, output)
+            delta_generator.maybe_generate(
+                driver_name, version, jar_artifact,
+                (previous_driver or {}).get("jar"), output, "",
+            )
             outputs.append(output)
 
         for platform, artifact in driver.get("native", {}).items():
@@ -104,6 +245,10 @@ def build_driver_zips(release_dir: Path) -> list[Path]:
             elif not output.is_file():
                 raise FileExistsError(f"Reusable native agent package is not a file: {output}")
             update_release_artifact(artifact, output)
+            delta_generator.maybe_generate(
+                driver_name, version, artifact,
+                (previous_driver or {}).get("native", {}).get(platform), output, f"-{platform}",
+            )
             outputs.append(output)
 
     registry_path.write_text(json.dumps(registry, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
@@ -113,7 +258,7 @@ def build_driver_zips(release_dir: Path) -> list[Path]:
 def remove_raw_driver_artifacts(release_dir: Path) -> list[Path]:
     removed: list[Path] = []
     for path in sorted(release_dir.glob("dbx-agent-*")):
-        if path.name.endswith(".tar.zst") or not path.is_file():
+        if path.name.endswith(".tar.zst") or path.name.endswith(".delta") or not path.is_file():
             continue
         path.unlink()
         removed.append(path)
@@ -124,9 +269,22 @@ def main() -> None:
     parser = argparse.ArgumentParser(description="Build tar.zst packages for individual DBX agents")
     parser.add_argument("release_dir", type=Path)
     parser.add_argument("--cleanup-sources", action="store_true")
+    parser.add_argument(
+        "--previous-registry",
+        type=Path,
+        default=None,
+        help=(
+            "agent-registry.json of the previous agents release; enables "
+            "incremental delta patches between consecutive driver versions"
+        ),
+    )
     args = parser.parse_args()
 
-    for path in build_driver_zips(args.release_dir):
+    previous_registry = None
+    if args.previous_registry and args.previous_registry.is_file():
+        previous_registry = json.loads(args.previous_registry.read_text(encoding="utf-8"))
+
+    for path in build_driver_zips(args.release_dir, previous_registry):
         print(f"Prepared {path.name} ({path.stat().st_size} bytes)")
     if args.cleanup_sources:
         for path in remove_raw_driver_artifacts(args.release_dir):

agents/scripts/driver_release_packages_test.py

@@ -1,6 +1,8 @@
 #!/usr/bin/env python3
+import hashlib
 import io
 import json
+import secrets
 import subprocess
 import tarfile
 import tempfile
@@ -466,5 +468,144 @@ class DriverReleasePackagesTest(unittest.TestCase):
                 self.assertIn(f"drivers/{kafka_filename}", archive.namelist())
 
 
+class DriverDeltaPatchTest(unittest.TestCase):
+    """Incremental `zstd --patch-from` deltas emitted by build_driver_zips."""
+
+    @staticmethod
+    def _incompressible_filler(size: int) -> bytes:
+        # Deterministic, incompressible shared payload: two packages built from
+        # it stay ~`size` bytes large while their version differences stay a
+        # few bytes, so the delta ratio reflects the small real change.
+        chunks = []
+        counter = 0
+        while sum(len(chunk) for chunk in chunks) < size:
+            chunks.append(hashlib.sha256(f"demo-filler-{counter}".encode()).digest())
+            counter += 1
+        return b"".join(chunks)[:size]
+
+    @staticmethod
+    def _write_release(release_dir: Path, version: str, jar_payload: bytes) -> None:
+        jar = release_dir / f"dbx-agent-demo-{version}.jar"
+        jar.write_bytes(jar_payload)
+        registry = {
+            "jres": {"21": {"version": "21", "platforms": {}}},
+            "drivers": {
+                "demo": {
+                    "version": version,
+                    "label": "Demo",
+                    "min_app_version": "0.6.0",
+                    "jre": "21",
+                    "jar": {"url": f"https://example.com/{jar.name}", "size": jar.stat().st_size},
+                }
+            },
+        }
+        (release_dir / "agent-registry.json").write_text(json.dumps(registry), encoding="utf-8")
+
+    @staticmethod
+    def _zstd_available() -> bool:
+        try:
+            subprocess.run(["zstd", "--version"], capture_output=True, check=True)
+            return True
+        except (OSError, subprocess.CalledProcessError):
+            return False
+
+    def _build_previous_release(self, release_dir: Path) -> None:
+        filler = self._incompressible_filler(96 * 1024)
+        jar_v1 = b"PK-demo-jar-v1\x00" + filler + b"\x00trailer-v1"
+        self._write_release(release_dir, "0.1.0", jar_v1)
+        build_driver_zips(release_dir)
+        registry = json.loads((release_dir / "agent-registry.json").read_text(encoding="utf-8"))
+        package = release_dir / "dbx-agent-demo-0.1.0.tar.zst"
+        # Point the previous registry at the local package so the generator
+        # fetches its base the same way production fetches the previous release.
+        registry["drivers"]["demo"]["jar"]["url"] = package.resolve().as_uri()
+        (release_dir / "previous-agent-registry.json").write_text(json.dumps(registry), encoding="utf-8")
+
+    def test_delta_patch_generated_and_registry_annotated(self) -> None:
+        if not self._zstd_available():
+            self.skipTest("zstd CLI is unavailable")
+        with tempfile.TemporaryDirectory() as temp_dir:
+            release_dir = Path(temp_dir)
+            self._build_previous_release(release_dir)
+
+            filler = self._incompressible_filler(96 * 1024)
+            jar_v2 = b"PK-demo-jar-v2\x00" + filler + b"\x00trailer-v2"
+            self._write_release(release_dir, "0.2.0", jar_v2)
+            previous = json.loads((release_dir / "previous-agent-registry.json").read_text(encoding="utf-8"))
+            build_driver_zips(release_dir, previous, delta_min_full_size=0)
+
+            package_v2 = release_dir / "dbx-agent-demo-0.2.0.tar.zst"
+            delta_path = release_dir / "dbx-agent-demo-0.1.0-to-0.2.0.tar.zst.delta"
+            self.assertTrue(delta_path.is_file(), "delta file must be written into the release dir")
+            self.assertLess(delta_path.stat().st_size, package_v2.stat().st_size)
+
+            registry = json.loads((release_dir / "agent-registry.json").read_text(encoding="utf-8"))
+            delta = registry["drivers"]["demo"]["jar"].get("delta")
+            self.assertIsNotNone(delta, "registry must carry the delta block")
+            self.assertEqual(delta["base_version"], "0.1.0")
+            self.assertEqual(delta["size"], delta_path.stat().st_size)
+            self.assertEqual(delta["sha256"], hashlib.sha256(delta_path.read_bytes()).hexdigest())
+            self.assertTrue(delta["url"].endswith(delta_path.name))
+            self.assertEqual(delta["url"].rpartition("/")[0], "https://example.com")
+
+            # The delta must reconstruct the exact full package from the base.
+            reconstructed = release_dir / "reconstructed.tar.zst"
+            subprocess.run(
+                [
+                    "zstd", "-q", "-d", "--patch-from",
+                    str(release_dir / "dbx-agent-demo-0.1.0.tar.zst"),
+                    str(delta_path), "-o", str(reconstructed), "--force",
+                ],
+                check=True,
+            )
+            self.assertEqual(reconstructed.read_bytes(), package_v2.read_bytes())
+
+    def test_delta_skipped_when_ratio_exceeds_threshold(self) -> None:
+        if not self._zstd_available():
+            self.skipTest("zstd CLI is unavailable")
+        with tempfile.TemporaryDirectory() as temp_dir:
+            release_dir = Path(temp_dir)
+            self._build_previous_release(release_dir)
+            # A completely unrelated payload produces a delta as large as the
+            # full package; the generator must drop it and leave no block.
+            self._write_release(release_dir, "0.2.0", secrets.token_bytes(96 * 1024))
+            previous = json.loads((release_dir / "previous-agent-registry.json").read_text(encoding="utf-8"))
+            build_driver_zips(release_dir, previous, delta_min_full_size=0, delta_max_ratio=0.5)
+
+            self.assertEqual(list(release_dir.glob("*.delta")), [])
+            registry = json.loads((release_dir / "agent-registry.json").read_text(encoding="utf-8"))
+            self.assertIsNone(registry["drivers"]["demo"]["jar"].get("delta"))
+
+    def test_delta_skipped_for_unchanged_version(self) -> None:
+        if not self._zstd_available():
+            self.skipTest("zstd CLI is unavailable")
+        with tempfile.TemporaryDirectory() as temp_dir:
+            release_dir = Path(temp_dir)
+            self._build_previous_release(release_dir)
+            # Same version in the previous registry: no delta, and the previous
+            # artifact must not even be fetched (its URL is intentionally dead).
+            previous = json.loads((release_dir / "previous-agent-registry.json").read_text(encoding="utf-8"))
+            previous["drivers"]["demo"]["jar"]["url"] = "https://example.invalid/no-such-package.tar.zst"
+            build_driver_zips(release_dir, previous, delta_min_full_size=0)
+            self.assertEqual(list(release_dir.glob("*.delta")), [])
+
+    def test_cleanup_keeps_delta_files(self) -> None:
+        with tempfile.TemporaryDirectory() as temp_dir:
+            release_dir = Path(temp_dir)
+            delta = release_dir / "dbx-agent-demo-0.1.0-to-0.2.0.tar.zst.delta"
+            delta.write_bytes(b"delta-frame")
+            raw = release_dir / "dbx-agent-demo-0.2.0.jar"
+            raw.write_bytes(b"raw")
+            removed = remove_raw_driver_artifacts(release_dir)
+            self.assertEqual(removed, [raw])
+            self.assertTrue(delta.is_file())
+
+    def test_release_uploads_delta_files_to_r2(self) -> None:
+        workflow = (Path(__file__).resolve().parents[2] / ".github/workflows/agents-release.yml").read_text(
+            encoding="utf-8"
+        )
+        self.assertIn('--include "dbx-agent-*.tar.zst.delta"', workflow)
+
+
 if __name__ == "__main__":
     unittest.main()

crates/dbx-core/tests/agent_service.rs

@@ -128,6 +128,7 @@ fn registry_with_driver(db_type: &str, version: &str, jre: &str) -> AgentRegistr
                 sha256: None,
                 size: 42,
                 format: None,
+                delta: None,
             }),
             native: std::collections::HashMap::new(),
         },
@@ -158,6 +159,7 @@ fn registry_with_native_driver(db_type: &str, version: &str, jre: &str) -> Agent
                 sha256: None,
                 size: 0,
                 format: None,
+                delta: None,
             }),
             native: [(
                 AgentManager::current_platform().to_string(),
@@ -166,6 +168,7 @@ fn registry_with_native_driver(db_type: &str, version: &str, jre: &str) -> Agent
                     sha256: None,
                     size: 42,
                     format: None,
+                    delta: None,
                 },
             )]
             .into_iter()
@@ -190,6 +193,7 @@ fn registry_with_foreign_platform_native_driver(db_type: &str, version: &str) ->
                     sha256: None,
                     size: 42,
                     format: None,
+                    delta: None,
                 },
             )
         })
@@ -209,6 +213,7 @@ fn registry_with_foreign_platform_native_driver(db_type: &str, version: &str) ->
                 sha256: None,
                 size: 0,
                 format: None,
+                delta: None,
             }),
             native,
         },

crates/dbx-driver-agent/src/agent_manager.rs

@@ -184,6 +184,42 @@ mod tests {
         }
     }
 
+    #[test]
+    fn registry_artifact_parses_optional_delta_and_stays_backward_compatible() {
+        let with_delta: ArtifactInfo = serde_json::from_str(
+            r#"{
+                "url": "https://example.com/dbx-agent-demo-0.2.0.tar.zst",
+                "sha256": "52da05589c140cdf5ceba54999422dfc50a1fd23ee19c9f26ee92869ad099d09",
+                "size": 71239014,
+                "delta": {
+                    "base_version": "0.1.73",
+                    "url": "https://example.com/dbx-agent-demo-0.1.73-to-0.2.0.tar.zst.delta",
+                    "sha256": "aa05555555c140cdf5feba54999422dfc50a1fd23ee19c9f26ee92869ad099d09",
+                    "size": 4432615
+                }
+            }"#,
+        )
+        .unwrap();
+        let delta = with_delta.delta.as_ref().expect("delta parsed");
+        assert_eq!(delta.base_version, "0.1.73");
+        assert_eq!(delta.size, 4_432_615);
+        assert!(delta.url.ends_with(".delta"));
+        // Round-trips through serialization for the offline export path.
+        let serialized = serde_json::to_value(&with_delta).unwrap();
+        assert!(serialized.get("delta").is_some());
+
+        // Registries published before delta support have no `delta` key.
+        let without_delta: ArtifactInfo = serde_json::from_str(
+            r#"{
+                "url": "https://example.com/dbx-agent-demo-0.1.73.tar.zst",
+                "sha256": "52da05589c140cdf5ceba54999422dfc50a1fd23ee19c9f26ee92869ad099d09",
+                "size": 71239014
+            }"#,
+        )
+        .unwrap();
+        assert!(without_delta.delta.is_none());
+    }
+
     #[test]
     fn cleanup_pending_jre_removes_stash_dirs_and_persists() {
         let manager = test_manager("pending-cleanup");
@@ -704,6 +740,25 @@ pub struct ArtifactInfo {
     pub size: u64,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub format: Option<ArtifactFormat>,
+    /// Optional incremental update payload: a zstd patch-from frame that
+    /// reconstructs this artifact from the artifact of `base_version`. Clients
+    /// without a matching retained base (or on any reconstruct failure) fall
+    /// back to downloading `url` in full.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub delta: Option<DeltaInfo>,
+}
+
+/// Describes the incremental update payload of an [ArtifactInfo].
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct DeltaInfo {
+    /// Version whose downloaded artifact the delta applies onto.
+    pub base_version: String,
+    /// URL of the delta frame (a `.delta` file on the same release as the full artifact).
+    pub url: String,
+    /// SHA-256 of the delta file itself.
+    pub sha256: String,
+    /// Size of the delta file in bytes.
+    pub size: u64,
 }
 
 #[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]

crates/dbx-driver-agent/src/agent_offline_export.rs

@@ -442,6 +442,7 @@ fn build_export_registry(am: &AgentManager, drivers: &[PreparedDriver], jres: &[
             sha256: Some(driver.sha256.clone()),
             size: driver.size,
             format: None,
+            delta: None,
         };
         let (jar, native) = match driver.kind {
             AgentOfflineArtifactKind::Jar => (Some(artifact), HashMap::new()),
@@ -473,6 +474,7 @@ fn build_export_registry(am: &AgentManager, drivers: &[PreparedDriver], jres: &[
                         sha256: Some(jre.sha256.clone()),
                         size: jre.size,
                         format: Some(ArtifactFormat::TarZstd),
+                        delta: None,
                     },
                 )]),
             },

crates/dbx-driver-agent/src/agent_service.rs

@@ -1665,18 +1665,14 @@ async fn install_sqlite_worker_from_registry(
             current,
             total_drivers,
         ));
-        download_with_progress(
+        obtain_driver_artifact(
             am,
             progress,
-            "driver",
             source,
-            &artifact.url,
-            &r2_path_with_cache_buster(&github_url_to_r2_path(&artifact.url, "driver"), &driver.version),
+            artifact,
+            db_type,
+            &driver.version,
             &download_path,
-            artifact.size,
-            artifact.sha256.as_deref(),
-            Some(CacheIdentity::Driver { db_type, version: &driver.version }),
-            Some(db_type),
             current,
             total_drivers,
             cancellations,
@@ -1798,18 +1794,14 @@ async fn install_agent_driver_from_registry(
         current,
         total_drivers,
     ));
-    download_with_progress(
+    obtain_driver_artifact(
         am,
         progress,
-        "driver",
         source,
-        &artifact.url,
-        &r2_path_with_cache_buster(&github_url_to_r2_path(&artifact.url, "driver"), &driver.version),
+        artifact,
+        db_type,
+        &driver.version,
         &download_path,
-        artifact.size,
-        artifact.sha256.as_deref(),
-        Some(CacheIdentity::Driver { db_type, version: &driver.version }),
-        Some(db_type),
         current,
         total_drivers,
         cancellations,
@@ -1821,6 +1813,9 @@ async fn install_agent_driver_from_registry(
         std::fs::remove_file(&download_path).ok();
         return Err(AGENT_DOWNLOAD_CANCELED_ERROR.to_string());
     }
+    // Retain the artifact bytes for future incremental updates before the
+    // install consumes (and deletes) the staged download.
+    retain_driver_delta_base(am, db_type, &driver.version, &download_path);
     stop_driver_processes_before_replacement(am, db_type).await?;
     install_downloaded_driver_artifact(
         &download_path,
@@ -2048,6 +2043,131 @@ fn agent_registry_driver<'a>(
     registry.drivers.get(db_type)
 }
 
+/// Downloads `artifact` for installation, preferring the registry's
+/// incremental delta when a matching base artifact was retained by a previous
+/// install of `db_type`. The staged full artifact lands at `dest` either way;
+/// every delta failure falls back to a full download.
+#[allow(clippy::too_many_arguments)]
+async fn obtain_driver_artifact(
+    am: &AgentManager,
+    progress: &impl Fn(AgentProgressEvent),
+    source: DownloadSource,
+    artifact: &crate::agent_manager::ArtifactInfo,
+    db_type: &str,
+    driver_version: &str,
+    dest: &Path,
+    current: Option<u32>,
+    total_drivers: Option<u32>,
+    cancellations: &[&AgentInstallCancellation],
+) -> Result<(), String> {
+    if let Some(delta) = artifact.delta.as_ref() {
+        let base = crate::driver_delta::find_delta_base(&am.download_cache_dir(), db_type, &delta.base_version);
+        if let Some(base) = base {
+            match try_delta_artifact(
+                am,
+                progress,
+                source,
+                artifact,
+                delta,
+                &base,
+                db_type,
+                driver_version,
+                dest,
+                current,
+                total_drivers,
+                cancellations,
+            )
+            .await
+            {
+                Ok(()) => {
+                    log::info!(
+                        "[driver-delta:{db_type}] applied incremental update {} -> {} (delta {} bytes of a {} byte artifact)",
+                        delta.base_version,
+                        driver_version,
+                        delta.size,
+                        artifact.size
+                    );
+                    return Ok(());
+                }
+                Err(err) => {
+                    log::warn!(
+                        "[driver-delta:{db_type}] incremental update failed, falling back to full download: {err}"
+                    );
+                    let _ = std::fs::remove_file(dest);
+                }
+            }
+        }
+    }
+    download_with_progress(
+        am,
+        progress,
+        "driver",
+        source,
+        &artifact.url,
+        &r2_path_with_cache_buster(&github_url_to_r2_path(&artifact.url, "driver"), driver_version),
+        dest,
+        artifact.size,
+        artifact.sha256.as_deref(),
+        Some(CacheIdentity::Driver { db_type, version: driver_version }),
+        Some(db_type),
+        current,
+        total_drivers,
+        cancellations,
+    )
+    .await
+}
+
+/// Downloads the delta frame and reconstructs the full artifact onto `dest`.
+/// The reconstructed bytes are verified against the full artifact's size and
+/// SHA-256; a mismatching (e.g. corrupt) base is caught here and reported as a
+/// plain error so the caller can fall back to a full download.
+#[allow(clippy::too_many_arguments)]
+async fn try_delta_artifact(
+    am: &AgentManager,
+    progress: &impl Fn(AgentProgressEvent),
+    source: DownloadSource,
+    artifact: &crate::agent_manager::ArtifactInfo,
+    delta: &crate::agent_manager::DeltaInfo,
+    base: &Path,
+    db_type: &str,
+    driver_version: &str,
+    dest: &Path,
+    current: Option<u32>,
+    total_drivers: Option<u32>,
+    cancellations: &[&AgentInstallCancellation],
+) -> Result<(), String> {
+    let file_name = dest
+        .file_name()
+        .and_then(|value| value.to_str())
+        .ok_or_else(|| "Driver artifact has no file name".to_string())?;
+    let delta_path = dest
+        .parent()
+        .ok_or_else(|| "Driver artifact has no parent directory".to_string())?
+        .join(format!(".{file_name}.delta"));
+    let cache_buster_version = format!("{driver_version}-from-{}", delta.base_version);
+    download_with_progress(
+        am,
+        progress,
+        "driver",
+        source,
+        &delta.url,
+        &r2_path_with_cache_buster(&github_url_to_r2_path(&delta.url, "driver"), &cache_buster_version),
+        &delta_path,
+        delta.size,
+        Some(delta.sha256.as_str()),
+        Some(CacheIdentity::Driver { db_type, version: &cache_buster_version }),
+        Some(db_type),
+        current,
+        total_drivers,
+        cancellations,
+    )
+    .await?;
+    let outcome = crate::driver_delta::apply_zstd_delta(base, &delta_path, dest)
+        .and_then(|()| validate_artifact_integrity(dest, artifact.size, artifact.sha256.as_deref()));
+    let _ = std::fs::remove_file(&delta_path);
+    outcome
+}
+
 #[allow(clippy::too_many_arguments)]
 async fn download_with_progress(
     am: &AgentManager,
@@ -2525,11 +2645,28 @@ fn prune_jre_download_cache(am: &AgentManager, jre_key: &str) -> Result<(), Stri
 }
 
 fn cleanup_driver_download_cache_after_success(am: &AgentManager, db_type: &str) {
-    if let Err(err) = prune_driver_download_cache(am, db_type) {
+    // Keep the retained delta base (future incremental updates read it back);
+    // drop the transient download entries for this driver.
+    let prefix = format!("driver-{}-", cache_file_token(db_type));
+    let base_prefix = format!("driver-{}-base-", cache_file_token(db_type));
+    if let Err(err) = remove_download_cache_entries(
+        am,
+        |name| name.starts_with(&prefix) && !name.starts_with(&base_prefix),
+        "cached driver download",
+    ) {
         log::warn!("Failed to clean cached download for {db_type}: {err}");
     }
 }
 
+/// Retains the just-installed artifact as the delta base for the next
+/// incremental update of `db_type`. Best-effort: a missing base only costs a
+/// full download later.
+fn retain_driver_delta_base(am: &AgentManager, db_type: &str, version: &str, downloaded: &Path) {
+    if let Err(err) = crate::driver_delta::retain_delta_base(&am.download_cache_dir(), db_type, version, downloaded) {
+        log::warn!("Failed to retain delta base for {db_type} {version}: {err}");
+    }
+}
+
 fn cleanup_jre_download_cache_after_success(am: &AgentManager, jre_key: &str) {
     if let Err(err) = prune_jre_download_cache(am, jre_key) {
         log::warn!("Failed to clean cached JRE download for {jre_key}: {err}");
@@ -4543,10 +4680,17 @@ mod agent_registry_install_tests {
                     sha256: None,
                     size: 0,
                     format: None,
+                    delta: None,
                 }),
                 native: [(
                     AgentManager::current_platform().to_string(),
-                    ArtifactInfo { url: native_url.to_string(), sha256: None, size: native_size, format: None },
+                    ArtifactInfo {
+                        url: native_url.to_string(),
+                        sha256: None,
+                        size: native_size,
+                        format: None,
+                        delta: None,
+                    },
                 )]
                 .into_iter()
                 .collect(),
@@ -4564,7 +4708,7 @@ mod agent_registry_install_tests {
                 label: db_type.to_string(),
                 min_app_version: "0.1.0".to_string(),
                 jre: DEFAULT_JRE_KEY.to_string(),
-                jar: Some(ArtifactInfo { url: url.to_string(), sha256: None, size, format: None }),
+                jar: Some(ArtifactInfo { url: url.to_string(), sha256: None, size, format: None, delta: None }),
                 native: std::collections::HashMap::new(),
             },
         );
@@ -4809,7 +4953,7 @@ mod agent_registry_install_tests {
                     version: version.to_string(),
                     platforms: [(
                         AgentManager::current_platform().to_string(),
-                        ArtifactInfo { url: url.to_string(), sha256: None, size, format: None },
+                        ArtifactInfo { url: url.to_string(), sha256: None, size, format: None, delta: None },
                     )]
                     .into_iter()
                     .collect(),
@@ -4951,6 +5095,7 @@ mod agent_registry_install_tests {
                 sha256: None,
                 size: 8,
                 format: None,
+                delta: None,
             },
         );
         cache_test_registry(registry).await;
@@ -5145,6 +5290,77 @@ mod agent_registry_install_tests {
         assert!(cache_path.exists());
     }
 
+    #[tokio::test]
+    async fn successful_delta_download_is_removed_by_driver_cache_cleanup() {
+        let manager = test_manager("delta-cache-cleanup");
+        let db_type = "demo";
+        let base_version = "0.1.0";
+        let driver_version = "0.2.0";
+        let delta_url = "https://example.invalid/demo.delta";
+        let rebuilt = b"reconstructed driver package";
+        let delta_bytes = zstd::stream::encode_all(rebuilt.as_slice(), 3).unwrap();
+        let delta_sha256 = format!("{:x}", Sha256::digest(&delta_bytes));
+        let artifact_sha256 = format!("{:x}", Sha256::digest(rebuilt));
+
+        let base_path = manager.download_cache_dir().join(crate::driver_delta::delta_base_file_name(
+            db_type,
+            base_version,
+            "base.tar.zst",
+        ));
+        std::fs::create_dir_all(base_path.parent().unwrap()).unwrap();
+        std::fs::write(&base_path, b"unused base for a standalone zstd frame").unwrap();
+
+        let dest = manager.driver_dir(db_type).join(".agent.tar.zst");
+        let delta_path = dest.parent().unwrap().join("..agent.tar.zst.delta");
+        let cache_version = format!("{driver_version}-from-{base_version}");
+        let cache_path = cached_download_path(
+            &manager,
+            delta_url,
+            delta_bytes.len() as u64,
+            Some(&delta_sha256),
+            Some(CacheIdentity::Driver { db_type, version: &cache_version }),
+            &delta_path,
+        );
+        std::fs::write(&cache_path, &delta_bytes).unwrap();
+
+        let artifact = ArtifactInfo {
+            url: "https://example.invalid/demo.tar.zst".to_string(),
+            sha256: Some(artifact_sha256),
+            size: rebuilt.len() as u64,
+            format: Some(ArtifactFormat::TarZstd),
+            delta: None,
+        };
+        let delta = crate::agent_manager::DeltaInfo {
+            base_version: base_version.to_string(),
+            url: delta_url.to_string(),
+            sha256: delta_sha256,
+            size: delta_bytes.len() as u64,
+        };
+
+        try_delta_artifact(
+            &manager,
+            &|_| {},
+            DownloadSource::Cnb,
+            &artifact,
+            &delta,
+            &base_path,
+            db_type,
+            driver_version,
+            &dest,
+            None,
+            None,
+            &[],
+        )
+        .await
+        .unwrap();
+
+        assert_eq!(std::fs::read(&dest).unwrap(), rebuilt);
+        assert!(cache_path.exists());
+        cleanup_driver_download_cache_after_success(&manager, db_type);
+        assert!(!cache_path.exists());
+        assert!(base_path.exists());
+    }
+
     #[tokio::test]
     async fn registry_install_replaces_hive_legacy_jar_with_native_driver() {
         let manager = test_manager("hive-native-replaces-legacy-jar");
@@ -5197,11 +5413,23 @@ mod agent_registry_install_tests {
         let mut native = std::collections::HashMap::new();
         native.insert(
             "linux-x64".to_string(),
-            ArtifactInfo { url: x64_url.to_string(), sha256: None, size: x64_bytes.len() as u64, format: None },
+            ArtifactInfo {
+                url: x64_url.to_string(),
+                sha256: None,
+                size: x64_bytes.len() as u64,
+                format: None,
+                delta: None,
+            },
         );
         native.insert(
             "linux-aarch64".to_string(),
-            ArtifactInfo { url: arm_url.to_string(), sha256: None, size: arm_bytes.len() as u64, format: None },
+            ArtifactInfo {
+                url: arm_url.to_string(),
+                sha256: None,
+                size: arm_bytes.len() as u64,
+                format: None,
+                delta: None,
+            },
         );
         let mut drivers = std::collections::HashMap::new();
         drivers.insert(
@@ -5216,6 +5444,7 @@ mod agent_registry_install_tests {
                     sha256: None,
                     size: 0,
                     format: None,
+                    delta: None,
                 }),
                 native,
             },
@@ -6387,6 +6616,7 @@ mod agent_registry_install_tests {
                             sha256: Some(sha256_bytes(&jre_bytes)),
                             size: jre_bytes.len() as u64,
                             format: None,
+                            delta: None,
                         },
                     )]
                     .into_iter()
@@ -6406,6 +6636,7 @@ mod agent_registry_install_tests {
                         sha256: Some(sha256_bytes(&jar_bytes)),
                         size: jar_bytes.len() as u64,
                         format: None,
+                        delta: None,
                     }),
                     native: std::collections::HashMap::new(),
                     jre: jre_key.to_string(),
@@ -6466,6 +6697,7 @@ mod agent_registry_install_tests {
                             sha256: Some(sha256_bytes(&unrelated_jre_bytes)),
                             size: unrelated_jre_bytes.len() as u64,
                             format: None,
+                            delta: None,
                         },
                     )]
                     .into_iter()
@@ -6485,6 +6717,7 @@ mod agent_registry_install_tests {
                         sha256: Some(sha256_bytes(&jar_bytes)),
                         size: jar_bytes.len() as u64,
                         format: None,
+                        delta: None,
                     }),
                     native: std::collections::HashMap::new(),
                     jre: "temurin-21".to_string(),
@@ -6527,6 +6760,7 @@ mod agent_registry_install_tests {
                             sha256: Some(sha256),
                             size,
                             format: None,
+                            delta: None,
                         },
                     )]
                     .into_iter()
@@ -6571,6 +6805,7 @@ mod agent_registry_install_tests {
                             sha256: Some(sha256_bytes(&jre_bytes)),
                             size: jre_bytes.len() as u64,
                             format: None,
+                            delta: None,
                         },
                     )]
                     .into_iter()
@@ -6590,6 +6825,7 @@ mod agent_registry_install_tests {
                         sha256: Some(sha256_bytes(&jar_bytes)),
                         size: jar_bytes.len() as u64,
                         format: None,
+                        delta: None,
                     }),
                     native: std::collections::HashMap::new(),
                     jre: jre_key.to_string(),
@@ -6666,6 +6902,7 @@ mod agent_registry_install_tests {
                         sha256: None,
                         size: 0,
                         format: None,
+                        delta: None,
                     },
                 )
             })

crates/dbx-driver-agent/src/driver_delta.rs

@@ -0,0 +1,331 @@
+//! Incremental (delta) driver artifact updates.
+//!
+//! A delta artifact is a zstd frame produced by compressing the new package
+//! with the previous package as a raw-content prefix (`zstd --patch-from`).
+//! Reconstructing the new package only needs the previously downloaded
+//! artifact — retained in the download cache as a "delta base" after every
+//! successful install — plus a small delta file. The reconstructed output is
+//! always verified against the full artifact's SHA-256 before it is used, and
+//! every failure path falls back to a full download, so a missing or corrupt
+//! base can never break an install.
+
+use std::path::{Path, PathBuf};
+
+use zstd::zstd_safe::{DCtx, DParameter, InBuffer, OutBuffer};
+
+/// `ZSTD_d_windowLogMax` cap for decoding patch-from frames. The window must
+/// cover the whole base artifact; 31 (2 GiB) is above any driver package we
+/// ship, and zstd only allocates what the frame actually needs.
+const DECODE_WINDOW_LOG_MAX: u32 = 31;
+
+/// Reconstructs the full artifact at `dest_path` by applying the delta frame
+/// `delta_path` onto the base artifact `base_path`.
+///
+/// Returns an error on any I/O or zstd failure; callers treat errors as a
+/// signal to fall back to a full download. If the base does not match the one
+/// the delta was produced from, decoding may still "succeed" with garbage
+/// output — the caller's SHA-256 check on the result is what catches that.
+pub fn apply_zstd_delta(base_path: &Path, delta_path: &Path, dest_path: &Path) -> Result<(), String> {
+    let base =
+        std::fs::read(base_path).map_err(|err| format!("Failed to read delta base {}: {err}", base_path.display()))?;
+    let delta = std::fs::read(delta_path)
+        .map_err(|err| format!("Failed to read delta artifact {}: {err}", delta_path.display()))?;
+
+    let mut output = Vec::new();
+    decode_with_prefix(&base, &delta, &mut output)?;
+
+    if let Some(parent) = dest_path.parent() {
+        std::fs::create_dir_all(parent).map_err(|err| format!("Failed to create delta output directory: {err}"))?;
+    }
+    std::fs::write(dest_path, &output)
+        .map_err(|err| format!("Failed to write reconstructed artifact {}: {err}", dest_path.display()))?;
+    Ok(())
+}
+
+fn decode_with_prefix(prefix: &[u8], frame: &[u8], output: &mut Vec<u8>) -> Result<(), String> {
+    let mut ctx = DCtx::create();
+    ctx.set_parameter(DParameter::WindowLogMax(DECODE_WINDOW_LOG_MAX))
+        .map_err(|code| format!("zstd decode window configuration failed: {code:?}"))?;
+    ctx.ref_prefix(prefix).map_err(|code| format!("zstd delta base registration failed: {code:?}"))?;
+
+    let mut input_pos = 0usize;
+    let mut out_scratch = vec![0_u8; DCtx::out_size().max(64 * 1024)];
+
+    loop {
+        let mut src = InBuffer::around(&frame[input_pos..]);
+        let mut dst = OutBuffer::around(&mut out_scratch);
+        let code =
+            ctx.decompress_stream(&mut dst, &mut src).map_err(|code| format!("zstd delta decode failed: {code:?}"))?;
+        output.extend_from_slice(dst.as_slice());
+        input_pos += src.pos();
+        if code == 0 && input_pos >= frame.len() {
+            return Ok(());
+        }
+        if code > 0 && src.pos() == 0 && input_pos >= frame.len() {
+            return Err("zstd delta frame is truncated".to_string());
+        }
+    }
+}
+
+/// File name of the retained delta base for `db_type` at `version` inside the
+/// download cache. The `driver-<db_type>-` prefix keeps bases aligned with the
+/// existing per-driver cache pruning (uninstall, manual cache clear), and the
+/// `base-<version>-` segment lets [find_delta_base] locate a base by version
+/// without knowing the full-artifact URL metadata of that release.
+pub fn delta_base_file_name(db_type: &str, version: &str, artifact_file_name: &str) -> String {
+    format!("driver-{}-base-{}-{}", cache_file_token(db_type), cache_file_token(version), artifact_file_name)
+}
+
+/// Retains `downloaded_artifact` as the delta base for future incremental
+/// updates of `db_type`, replacing any base retained for an older version.
+///
+/// The base is copied (not moved) so a caller that still needs the downloaded
+/// artifact for unpacking keeps working; the transient cache entries for the
+/// driver are pruned right after by the regular install cleanup, which is
+/// taught to preserve the base returned here.
+pub fn retain_delta_base(
+    cache_dir: &Path,
+    db_type: &str,
+    version: &str,
+    downloaded_artifact: &Path,
+) -> Result<PathBuf, String> {
+    let artifact_file_name = downloaded_artifact
+        .file_name()
+        .and_then(|name| name.to_str())
+        .ok_or_else(|| "Delta base artifact has no file name".to_string())?;
+    let base_name = delta_base_file_name(db_type, version, artifact_file_name);
+    let base_path = cache_dir.join(&base_name);
+
+    // Drop bases from other versions of this driver first so at most one base
+    // per driver is retained at any time (bounded disk usage).
+    let prefix = format!("driver-{}-base-", cache_file_token(db_type));
+    if let Ok(entries) = std::fs::read_dir(cache_dir) {
+        for entry in entries.flatten() {
+            let name = entry.file_name();
+            let Some(name) = name.to_str() else { continue };
+            if name.starts_with(&prefix) && name != base_name {
+                let _ = std::fs::remove_file(entry.path());
+            }
+        }
+    }
+
+    if let Some(parent) = base_path.parent() {
+        std::fs::create_dir_all(parent).map_err(|err| format!("Failed to create delta base directory: {err}"))?;
+    }
+    // Windows cannot rename onto an existing file; the version-scoped name
+    // makes collisions unlikely, but remove first to stay atomic-safe.
+    let _ = std::fs::remove_file(&base_path);
+    std::fs::copy(downloaded_artifact, &base_path)
+        .map_err(|err| format!("Failed to retain delta base {}: {err}", base_path.display()))?;
+    Ok(base_path)
+}
+
+/// Locates the retained delta base for `db_type` whose version matches
+/// `base_version`, if one survives in the download cache.
+pub fn find_delta_base(cache_dir: &Path, db_type: &str, base_version: &str) -> Option<PathBuf> {
+    let prefix = format!("driver-{}-base-{}-", cache_file_token(db_type), cache_file_token(base_version));
+    let entries = std::fs::read_dir(cache_dir).ok()?;
+    let mut candidates: Vec<PathBuf> = entries
+        .flatten()
+        .filter(|entry| entry.file_name().to_str().is_some_and(|name| name.starts_with(&prefix)))
+        .map(|entry| entry.path())
+        .collect();
+    candidates.sort();
+    candidates.into_iter().find(|path| path.is_file())
+}
+
+/// Reuses the same sanitizing rules as the driver download cache naming in
+/// `agent_service` so bases and transient entries share one namespace.
+fn cache_file_token(value: &str) -> String {
+    let token = value
+        .chars()
+        .map(|ch| if ch.is_ascii_alphanumeric() || ch == '-' || ch == '_' || ch == '.' { ch } else { '-' })
+        .collect::<String>()
+        .trim_matches('-')
+        .to_string();
+    if token.is_empty() {
+        "unknown".to_string()
+    } else {
+        token
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use zstd::zstd_safe::{CCtx, CParameter};
+
+    fn compress_with_prefix(prefix: &[u8], content: &[u8]) -> Vec<u8> {
+        let mut ctx = CCtx::create();
+        // Mirror `zstd --patch-from`: a window that covers the whole prefix so
+        // matches may reference any earlier offset of the base artifact.
+        let window_log = (usize::BITS - prefix.len().leading_zeros()).saturating_sub(1);
+        ctx.set_parameter(CParameter::WindowLog(window_log.max(10))).unwrap();
+        ctx.set_parameter(CParameter::CompressionLevel(3)).unwrap();
+        ctx.ref_prefix(prefix).unwrap();
+
+        let mut src = InBuffer::around(content);
+        let mut compressed = Vec::new();
+        let mut out_scratch = vec![0_u8; CCtx::out_size()];
+        while src.pos() < content.len() {
+            let mut dst = OutBuffer::around(&mut out_scratch);
+            ctx.compress_stream(&mut dst, &mut src).unwrap();
+            compressed.extend_from_slice(dst.as_slice());
+        }
+        loop {
+            let mut dst = OutBuffer::around(&mut out_scratch);
+            let remaining = ctx.end_stream(&mut dst).unwrap();
+            compressed.extend_from_slice(dst.as_slice());
+            if remaining == 0 {
+                break;
+            }
+        }
+        compressed
+    }
+
+    fn temp_dir(name: &str) -> PathBuf {
+        let dir = std::env::temp_dir().join(format!("dbx-driver-delta-{name}-{}", uuid::Uuid::new_v4()));
+        std::fs::create_dir_all(&dir).unwrap();
+        dir
+    }
+
+    fn synthetic_artifact(seed: &str, size: usize, mutate_from: usize) -> Vec<u8> {
+        // Deterministic pseudo-content with large shared regions, so prefix
+        // matching has real work to do instead of compressing from scratch.
+        let mut data = Vec::with_capacity(size);
+        let mut state: u64 = seed.bytes().map(|b| b as u64).sum();
+        for _ in 0..size {
+            state = state.wrapping_mul(6364136223846793005).wrapping_add(1442695040888963407);
+            data.push((state >> 33) as u8);
+        }
+        if mutate_from < size {
+            for (i, byte) in b"DELTA-MUTATED-PAYLOAD".iter().enumerate() {
+                let at = mutate_from + i;
+                if at < size {
+                    data[at] = *byte;
+                }
+            }
+        }
+        data
+    }
+
+    #[test]
+    fn delta_round_trip_reconstructs_new_artifact_bit_for_bit() {
+        let old = synthetic_artifact("old", 512 * 1024, usize::MAX);
+        let new = synthetic_artifact("old", 512 * 1024, 300 * 1024);
+        let frame = compress_with_prefix(&old, &new);
+        assert!(frame.len() < new.len(), "delta should be smaller than the full artifact");
+
+        let dir = temp_dir("roundtrip");
+        let base = dir.join("base.tar.zst");
+        let delta = dir.join("update.delta");
+        let out = dir.join("reconstructed.tar.zst");
+        std::fs::write(&base, &old).unwrap();
+        std::fs::write(&delta, &frame).unwrap();
+
+        apply_zstd_delta(&base, &delta, &out).unwrap();
+        assert_eq!(std::fs::read(&out).unwrap(), new);
+    }
+
+    #[test]
+    fn wrong_base_reconstructs_garbage_instead_of_succeeding_cleanly() {
+        // A base that does not match the delta's production base may decode
+        // without a zstd error; the output must then differ from the expected
+        // artifact, which is exactly what the caller's SHA-256 check detects.
+        let old = synthetic_artifact("old", 256 * 1024, usize::MAX);
+        let new = synthetic_artifact("old", 256 * 1024, 128 * 1024);
+        let frame = compress_with_prefix(&old, &new);
+        let other_base = synthetic_artifact("other", 256 * 1024, usize::MAX);
+
+        let dir = temp_dir("wrong-base");
+        let base = dir.join("base.tar.zst");
+        let delta = dir.join("update.delta");
+        let out = dir.join("reconstructed.tar.zst");
+        std::fs::write(&base, &other_base).unwrap();
+        std::fs::write(&delta, &frame).unwrap();
+
+        if apply_zstd_delta(&base, &delta, &out).is_ok() {
+            let reconstructed = std::fs::read(&out).unwrap();
+            assert_ne!(reconstructed, new, "mismatched base must not reproduce the target");
+        }
+    }
+
+    #[test]
+    fn truncated_delta_is_rejected() {
+        let old = synthetic_artifact("old", 128 * 1024, usize::MAX);
+        let new = synthetic_artifact("old", 128 * 1024, 64 * 1024);
+        let frame = compress_with_prefix(&old, &new);
+
+        let dir = temp_dir("truncated");
+        let base = dir.join("base.tar.zst");
+        let delta = dir.join("update.delta");
+        let out = dir.join("reconstructed.tar.zst");
+        std::fs::write(&base, &old).unwrap();
+        std::fs::write(&delta, &frame[..frame.len() / 2]).unwrap();
+
+        assert!(apply_zstd_delta(&base, &delta, &out).is_err());
+    }
+
+    #[test]
+    fn retained_base_is_found_by_version_and_replaced_by_newer_base() {
+        let dir = temp_dir("retain");
+        let artifact_v1 = dir.join("dbx-agent-demo-0.1.1.tar.zst");
+        let artifact_v2 = dir.join("dbx-agent-demo-0.1.2.tar.zst");
+        std::fs::write(&artifact_v1, b"v1-bytes").unwrap();
+        std::fs::write(&artifact_v2, b"v2-bytes").unwrap();
+
+        let cache = dir.join("cache");
+        std::fs::create_dir_all(&cache).unwrap();
+
+        retain_delta_base(&cache, "demo", "0.1.1", &artifact_v1).unwrap();
+        let found = find_delta_base(&cache, "demo", "0.1.1").expect("base for 0.1.1 retained");
+        assert_eq!(std::fs::read(&found).unwrap(), b"v1-bytes");
+        assert!(find_delta_base(&cache, "demo", "0.1.2").is_none());
+
+        // Retaining the next version drops the previous base: one base per driver.
+        retain_delta_base(&cache, "demo", "0.1.2", &artifact_v2).unwrap();
+        assert!(find_delta_base(&cache, "demo", "0.1.1").is_none());
+        let found = find_delta_base(&cache, "demo", "0.1.2").expect("base for 0.1.2 retained");
+        assert_eq!(std::fs::read(&found).unwrap(), b"v2-bytes");
+    }
+
+    /// Real-world end-to-end check against published release artifacts.
+    ///
+    /// Ignored by default (CI has no network guarantees and the artifacts are
+    /// ~68 MB); run manually after downloading two consecutive versions of a
+    /// driver package, e.g. the snowflake pair from `agents-v0.2.129` and
+    /// `agents-v0.2.130`:
+    ///
+    /// ```text
+    /// DRIVER_DELTA_E2E_BASE=<old.tar.zst> DRIVER_DELTA_E2E_DELTA=<old-to-new.tar.zst.delta> \
+    ///   DRIVER_DELTA_E2E_EXPECTED_SHA256=<sha256 of new.tar.zst> \
+    ///   cargo nextest run -p dbx-driver-agent e2e_real --run-ignored=only
+    /// ```
+    #[test]
+    #[ignore = "requires manually downloaded release artifacts via DRIVER_DELTA_E2E_* env vars"]
+    fn e2e_real_artifact_delta_reconstruction() {
+        let Ok(base) = std::env::var("DRIVER_DELTA_E2E_BASE") else {
+            panic!("DRIVER_DELTA_E2E_BASE (previous full artifact) is required");
+        };
+        let Ok(delta) = std::env::var("DRIVER_DELTA_E2E_DELTA") else {
+            panic!("DRIVER_DELTA_E2E_DELTA (delta frame) is required");
+        };
+        let expected_sha = std::env::var("DRIVER_DELTA_E2E_EXPECTED_SHA256")
+            .expect("DRIVER_DELTA_E2E_EXPECTED_SHA256 (sha256 of the new full artifact) is required");
+
+        let dir = temp_dir("e2e-real");
+        let out = dir.join("reconstructed.tar.zst");
+        apply_zstd_delta(std::path::Path::new(&base), std::path::Path::new(&delta), &out)
+            .expect("real delta must apply");
+
+        let actual_sha = {
+            use sha2::Digest;
+            let bytes = std::fs::read(&out).unwrap();
+            format!("{:x}", sha2::Sha256::digest(&bytes))
+        };
+        assert_eq!(
+            actual_sha, expected_sha,
+            "reconstructed artifact must match the published full artifact bit-for-bit"
+        );
+    }
+}

crates/dbx-driver-agent/src/lib.rs

@@ -15,6 +15,7 @@ pub mod agent_runtime;
 pub mod agent_service;
 pub mod backend_error;
 pub mod database_capabilities;
+pub mod driver_delta;
 pub mod oracle_oci;
 
 pub mod db {