From ad32874bc6fd2b33e591ec4ba4af6a07f9be3409 Mon Sep 17 00:00:00 2001
From: amarcozzi <anthonymarcozzi@gmail.com>
Date: Mon, 15 Jun 2026 22:52:06 -0600
Subject: [PATCH] Replace inventory in-place full rewrite with partial
 partition rewrite

In-place modifications and treatments rewrote the entire Parquet dataset
via a delete + copy-back staging swap, regardless of how many rows the
operation actually touched. Rewrite only the partitions whose content
changes.

standgen storage:
- write_changed_partitions(id, transform): read every partition in one
  concurrent gcsfs cat, apply the per-partition transform, and write back
  only the partitions where not new.equals(old) in one concurrent pipe.
  Leaves _metadata untouched (the file set is unchanged; readers re-read
  each footer). Replaces save_parquet_replace.
- write_full_partitions(id, df): materialized full rewrite for the one
  case that re-partitions globally; drops the now-wrong _metadata /
  _common_metadata and stale part files so readers list fresh.

Modifications route every mod through write_changed_partitions. Treatments
route by per-partition expressibility, not metric: diameter and directional
basal-area thins (the latter reduced to a single precomputed diameter cutoff
that reproduces the fastfuels-core thinner exactly) rewrite only changed
partitions, including polygon-scoped thins; only proportional basal-area
(random whole-stand removal) falls back to write_full_partitions.

API metadata/data endpoints read num_partitions / total_rows / columns from
the dask DataFrame (per-file footers) instead of the aggregated _metadata,
so counts stay correct after a partial rewrite leaves _metadata stale. Drops
the per-partition row-count list from the /data metadata response.

Adds an end-to-end benchmark (benchmarks/bench_inplace_inventory.py) that
times in-place modifications against the deployed API + standgen pipeline.

Closes #355
---
 .../api/api/resources/inventories/cache.py    |  73 ++-
 .../api/api/resources/inventories/router.py   |  18 +-
 .../api/api/resources/inventories/schema.py   |   6 -
 services/api/benchmarks/.gitignore            |   2 +
 .../api/benchmarks/bench_inplace_inventory.py | 428 ++++++++++++++++++
 .../resources/inventories/test_data_cache.py  |  74 ++-
 .../resources/inventories/test_data_router.py |   6 +-
 .../resources/inventories/test_schema.py      |  19 -
 .../standgen/handlers/modifications.py        |  19 +-
 .../standgen/standgen/handlers/treatments.py  |  51 ++-
 services/standgen/standgen/storage.py         | 140 +++---
 services/standgen/standgen/treatments.py      | 129 ++++++
 .../tests/handlers/test_modifications.py      | 101 ++---
 .../tests/handlers/test_treatments.py         | 219 +++++----
 services/standgen/tests/test_storage.py       | 101 ++++-
 15 files changed, 1062 insertions(+), 324 deletions(-)
 create mode 100644 services/api/benchmarks/.gitignore
 create mode 100644 services/api/benchmarks/bench_inplace_inventory.py

diff --git a/services/api/api/resources/inventories/cache.py b/services/api/api/resources/inventories/cache.py
index 78c9db70..7cc6433a 100644
--- a/services/api/api/resources/inventories/cache.py
+++ b/services/api/api/resources/inventories/cache.py
@@ -1,60 +1,41 @@
 """
 api/v2/resources/inventories/cache.py
 
-Cached async parquet metadata access for inventory data streaming.
+Cached async access to an inventory's partitioned-Parquet metadata and data.
+
+Counts come from the dask DataFrame, which reads each partition's own footer (and
+ignores any aggregated ``_metadata``). This stays correct after standgen rewrites
+only the partitions a modification changed — an in-place overwrite leaves the
+``_metadata`` aggregate stale, but ``len(ddf)`` does not trust it.
 """
 
 import asyncio
 from dataclasses import dataclass
 
+import dask.dataframe as dd
 import pandas as pd
-import pyarrow.parquet as pq
 from ring import lru
 
 from lib.config import INVENTORIES_BUCKET
 
 
-@dataclass
-class PartitionMeta:
-    index: int
-    num_rows: int
-    path: str
-
-
 @dataclass
 class InventoryMeta:
     num_partitions: int
     total_rows: int
     columns: list[str]
-    partitions: list[PartitionMeta]
-
-
-def _read_metadata_sync(inventory_id: str) -> InventoryMeta:
-    metadata_path = f"gs://{INVENTORIES_BUCKET}/{inventory_id}/_metadata"
-    pf = pq.ParquetFile(metadata_path)
-    return _parse_metadata(pf.metadata)
 
 
-def _parse_metadata(metadata: pq.FileMetaData) -> InventoryMeta:
-    partitions_by_path: dict[str, int] = {}
-    for i in range(metadata.num_row_groups):
-        rg = metadata.row_group(i)
-        path = rg.column(0).file_path
-        partitions_by_path[path] = partitions_by_path.get(path, 0) + rg.num_rows
+def _inventory_path(inventory_id: str) -> str:
+    return f"gs://{INVENTORIES_BUCKET}/{inventory_id}"
 
-    columns = metadata.schema.to_arrow_schema().names
-
-    sorted_paths = sorted(partitions_by_path.keys())
-    partitions = [
-        PartitionMeta(index=idx, num_rows=partitions_by_path[path], path=path)
-        for idx, path in enumerate(sorted_paths)
-    ]
 
+def _read_metadata_sync(inventory_id: str) -> InventoryMeta:
+    ddf = dd.read_parquet(_inventory_path(inventory_id))
     return InventoryMeta(
-        num_partitions=len(partitions),
-        total_rows=sum(p.num_rows for p in partitions),
-        columns=columns,
-        partitions=partitions,
+        num_partitions=ddf.npartitions,
+        total_rows=len(ddf),  # footer-level row count; ignores stale _metadata
+        columns=list(ddf.columns),
     )
 
 
@@ -62,23 +43,29 @@ def _parse_metadata(metadata: pq.FileMetaData) -> InventoryMeta:
 async def get_inventory_metadata(
     inventory_id: str, checksum: str | None
 ) -> InventoryMeta:
-    """Read an inventory's partitioned-Parquet metadata, cached per content.
+    """Read an inventory's partition metadata, cached per content.
 
     ``checksum`` participates only in the cache key (the body ignores it). An
-    in-place modification (POST .../modifications) rewrites the data under the
-    same ``inventory_id`` and re-assigns the inventory's ``checksum``, so a new
-    checksum bypasses the stale entry. Without it, this LRU would keep serving
-    pre-modification partition paths and row counts. ``None`` (legacy
-    inventories without a checksum) is a valid, stable key — such inventories
-    predate in-place edits, and any in-place edit assigns a fresh checksum.
+    in-place modification rewrites the changed partitions under the same
+    ``inventory_id`` and re-assigns the inventory's ``checksum``, so a new
+    checksum bypasses the stale entry. ``None`` (legacy inventories without a
+    checksum) is a valid, stable key.
     """
     return await asyncio.to_thread(_read_metadata_sync, inventory_id)
 
 
+def _read_partition_sync(
+    inventory_id: str, partition_index: int, columns: list[str] | None
+) -> pd.DataFrame:
+    ddf = dd.read_parquet(_inventory_path(inventory_id), columns=columns)
+    return ddf.partitions[partition_index].compute().reset_index(drop=True)
+
+
 async def read_partition(
     inventory_id: str,
-    partition_path: str,
+    partition_index: int,
     columns: list[str] | None = None,
 ) -> pd.DataFrame:
-    path = f"gs://{INVENTORIES_BUCKET}/{inventory_id}/{partition_path}"
-    return await asyncio.to_thread(pd.read_parquet, path, columns=columns)
+    return await asyncio.to_thread(
+        _read_partition_sync, inventory_id, partition_index, columns
+    )
diff --git a/services/api/api/resources/inventories/router.py b/services/api/api/resources/inventories/router.py
index cbf3a641..e8332296 100644
--- a/services/api/api/resources/inventories/router.py
+++ b/services/api/api/resources/inventories/router.py
@@ -45,7 +45,6 @@
     InventoryDataMetadata,
     InventoryDataResponse,
     InventoryJsonOrientation,
-    InventoryPartitionInfo,
     InventorySortField,
     InventoryType,
     ListInventoriesResponse,
@@ -599,9 +598,9 @@ async def get_inventory_data_metadata(
     """
     # Get Inventory Data Metadata
 
-    Returns partition count, total rows, per-partition row counts, and column
-    names for a completed inventory. Reads only the `_metadata` file from GCS
-    (cached after first access).
+    Returns the partition count, total row count, and column names for a
+    completed inventory (cached after first access). To get a single partition's
+    rows — and its row count — fetch it from the partition data endpoint below.
 
     ## Path Parameters
 
@@ -611,10 +610,10 @@ async def get_inventory_data_metadata(
     ## Response
 
     - **inventory_id**: The inventory ID.
-    - **num_partitions**: Number of Parquet partitions.
+    - **num_partitions**: Number of Parquet partitions (addressable as `0` to
+      `num_partitions - 1` via the partition data endpoint).
     - **total_rows**: Total row count across all partitions.
     - **columns**: List of column names.
-    - **partitions**: Per-partition index and row count.
 
     ## Error Responses
 
@@ -647,10 +646,6 @@ async def get_inventory_data_metadata(
         num_partitions=meta.num_partitions,
         total_rows=meta.total_rows,
         columns=meta.columns,
-        partitions=[
-            InventoryPartitionInfo(index=p.index, num_rows=p.num_rows)
-            for p in meta.partitions
-        ],
     )
 
 
@@ -748,8 +743,7 @@ async def get_inventory_data(
                 detail=(f"Columns not found: {missing}. Available: {meta.columns}"),
             )
 
-    partition = meta.partitions[partition_index]
-    df = await read_partition(inventory_id, partition.path, columns=requested_columns)
+    df = await read_partition(inventory_id, partition_index, columns=requested_columns)
 
     col_names = list(df.columns)
 
diff --git a/services/api/api/resources/inventories/schema.py b/services/api/api/resources/inventories/schema.py
index 42c1467e..dc58d6a3 100644
--- a/services/api/api/resources/inventories/schema.py
+++ b/services/api/api/resources/inventories/schema.py
@@ -211,17 +211,11 @@ class InventoryJsonOrientation(StrEnum):
     records = "records"
 
 
-class InventoryPartitionInfo(BaseModel):
-    index: int
-    num_rows: int
-
-
 class InventoryDataMetadata(BaseModel):
     inventory_id: str
     num_partitions: int
     total_rows: int
     columns: list[str]
-    partitions: list[InventoryPartitionInfo]
 
 
 class InventoryDataResponse(BaseModel):
diff --git a/services/api/benchmarks/.gitignore b/services/api/benchmarks/.gitignore
new file mode 100644
index 00000000..e6d76c8f
--- /dev/null
+++ b/services/api/benchmarks/.gitignore
@@ -0,0 +1,2 @@
+# Benchmark result files (timestamped, carry run-specific resource IDs).
+bench_inplace_*.json
diff --git a/services/api/benchmarks/bench_inplace_inventory.py b/services/api/benchmarks/bench_inplace_inventory.py
new file mode 100644
index 00000000..ada75523
--- /dev/null
+++ b/services/api/benchmarks/bench_inplace_inventory.py
@@ -0,0 +1,428 @@
+"""
+End-to-end timing benchmark for in-place inventory modifications.
+
+Drives the *deployed* API + standgen pipeline (not a unit/integration test) to
+measure how long an in-place ``POST .../{inventory_id}/modifications`` takes to
+complete, end to end. Used to A/B the storage rewrite in #355: run once against
+the currently-deployed full-rewrite standgen, then again after deploying the
+partial-rewrite update, and compare.
+
+What it does:
+  1. Creates a sized domain, a TreeMap PIM grid, and a PIM tree inventory
+     (deterministic by ``--seed`` so before/after runs operate on identical data).
+  2. Reads the inventory's partition count / row count for context.
+  3. Times two modification scenarios, each repeated ``--reps`` times
+     (the first rep per scenario is discarded as a cold-start warmup):
+       - ``scoped``: multiply tree height inside a sub-region of the domain.
+         Touches only the partitions that region overlaps -> the partial-rewrite
+         win shows up here.
+       - ``global``: multiply every tree's height. Touches all partitions ->
+         a no-regression check (partial == full when everything changes).
+     A multiply (not a remove) is used so every rep does the same work on a
+     stable partition layout (no rows are dropped between reps).
+  4. Writes results to ``bench_inplace_<label>.json`` and prints a summary.
+  5. Deletes the domain/grid/inventory (unless ``--keep``).
+
+Usage (from services/api, so the API venv + repo-root .env are picked up):
+
+    uv run python benchmarks/bench_inplace_inventory.py --label before
+    # ... merge #355, deploy standgen-v2-prod ...
+    uv run python benchmarks/bench_inplace_inventory.py --label after
+
+Auth/config come from the repo-root ``.env`` (``TEST_API_KEY``). The base URL
+defaults to api-v2-prod; override with ``--base-url``.
+"""
+
+import argparse
+import json
+import math
+import statistics
+import time
+import uuid
+from pathlib import Path
+
+from dotenv import dotenv_values
+from httpx import Client
+
+PROD_BASE_URL = "https://api-v2-prod-nyvjyh5ywa-uw.a.run.app"
+
+# Blue Mountain Recreation Area center (near Missoula, MT) — same locale as the
+# canonical test domain, just sized up to yield a multi-partition inventory.
+CENTER_LON = -114.1038
+CENTER_LAT = 46.8287
+
+# services/api/benchmarks/bench_inplace_inventory.py -> repo root is 3 levels up.
+REPO_ROOT = Path(__file__).resolve().parents[3]
+
+
+def log(msg: str) -> None:
+    print(f"[{time.strftime('%H:%M:%S')}] {msg}", flush=True)
+
+
+def km_to_deg(center_lat: float, dx_km: float, dy_km: float) -> tuple[float, float]:
+    """Convert east/north kilometer extents to (dlon, dlat) degrees at a latitude."""
+    dlat = dy_km / 111.32
+    dlon = dx_km / (111.32 * math.cos(math.radians(center_lat)))
+    return dlon, dlat
+
+
+def square_bbox(half_km: float) -> tuple[float, float, float, float]:
+    """(min_lon, min_lat, max_lon, max_lat) of a square domain centered on the site."""
+    dlon, dlat = km_to_deg(CENTER_LAT, half_km, half_km)
+    return (
+        CENTER_LON - dlon,
+        CENTER_LAT - dlat,
+        CENTER_LON + dlon,
+        CENTER_LAT + dlat,
+    )
+
+
+def polygon_feature_collection(
+    bbox: tuple[float, float, float, float], name: str
+) -> dict:
+    min_lon, min_lat, max_lon, max_lat = bbox
+    ring = [
+        [min_lon, min_lat],
+        [max_lon, min_lat],
+        [max_lon, max_lat],
+        [min_lon, max_lat],
+        [min_lon, min_lat],
+    ]
+    return {
+        "type": "FeatureCollection",
+        "features": [
+            {
+                "type": "Feature",
+                "properties": {},
+                "geometry": {"type": "Polygon", "coordinates": [ring]},
+            }
+        ],
+        "name": name,
+        "description": "Sized domain for in-place modification benchmarking.",
+    }
+
+
+def scoped_strip_geometry(bbox: tuple[float, float, float, float], frac: float) -> dict:
+    """A full-width strip across the southern ``frac`` of the domain.
+
+    A strip (vs. a tiny box) is guaranteed to contain trees while still covering
+    only a fraction of the inventory's partitions, so the scoped scenario does
+    real, repeatable work without depending on lucky tree placement.
+    """
+    min_lon, min_lat, max_lon, max_lat = bbox
+    strip_top = min_lat + (max_lat - min_lat) * frac
+    ring = [
+        [min_lon, min_lat],
+        [max_lon, min_lat],
+        [max_lon, strip_top],
+        [min_lon, strip_top],
+        [min_lon, min_lat],
+    ]
+    return {
+        "type": "Polygon",
+        "coordinates": [ring],
+        # Carried alongside so the caller can attach the CRS to the condition.
+    }
+
+
+def scenarios(
+    bbox: tuple[float, float, float, float], frac: float, factor: float
+) -> dict[str, dict]:
+    """Return {name: modifications-request-body} for each timed scenario."""
+    wgs84 = {"type": "name", "properties": {"name": "EPSG:4326"}}
+    strip = scoped_strip_geometry(bbox, frac)
+    return {
+        "scoped": {
+            "modifications": [
+                {
+                    "conditions": {
+                        "source": "geometry",
+                        "operator": "within",
+                        "geometry": strip,
+                        "crs": wgs84,
+                    },
+                    "actions": {
+                        "attribute": "height",
+                        "modifier": "multiply",
+                        "value": factor,
+                    },
+                }
+            ]
+        },
+        "global": {
+            "modifications": [
+                {
+                    "conditions": {"attribute": "height", "operator": "gt", "value": 0},
+                    "actions": {
+                        "attribute": "height",
+                        "modifier": "multiply",
+                        "value": factor,
+                    },
+                }
+            ]
+        },
+    }
+
+
+class Bench:
+    def __init__(self, client: Client, poll_interval: float):
+        self.client = client
+        self.poll_interval = poll_interval
+
+    def poll(self, domain_id: str, kind: str, rid: str, timeout: float) -> dict:
+        """Poll a resource until it reaches a terminal status; return its doc."""
+        url = f"/domains/{domain_id}/{kind}/{rid}"
+        start = time.perf_counter()
+        while True:
+            r = self.client.get(url)
+            r.raise_for_status()
+            doc = r.json()
+            status = doc.get("status")
+            if status == "completed":
+                return doc
+            if status == "failed":
+                err = doc.get("error") or {}
+                raise RuntimeError(
+                    f"{kind}/{rid} failed: {err.get('code')} - {err.get('message')}"
+                )
+            if time.perf_counter() - start > timeout:
+                raise TimeoutError(
+                    f"{kind}/{rid} did not complete within {timeout}s "
+                    f"(last status={status}, progress={doc.get('progress')})"
+                )
+            time.sleep(self.poll_interval)
+
+    def create_domain(self, body: dict) -> str:
+        r = self.client.post("/domains", json=body, timeout=120.0)
+        r.raise_for_status()
+        return r.json()["id"]
+
+    def create_and_wait(
+        self, domain_id: str, kind: str, endpoint: str, body: dict, timeout: float
+    ) -> dict:
+        url = f"/domains/{domain_id}{endpoint}"
+        r = self.client.post(url, json=body, timeout=120.0)
+        if r.status_code != 201:
+            raise RuntimeError(f"POST {url} -> {r.status_code}: {r.text}")
+        rid = r.json()["id"]
+        log(f"  created {kind} {rid}, polling for completion...")
+        return self.poll(domain_id, kind, rid, timeout)
+
+    def time_modification(
+        self, domain_id: str, inventory_id: str, body: dict, timeout: float
+    ) -> float:
+        """POST a modification and block until the inventory is completed again.
+
+        Returns wall-clock seconds from POST to observed completion. The POST
+        sets status=pending synchronously (in a Firestore transaction) before
+        returning, so polling to completed can't observe a stale prior state.
+        """
+        url = f"/domains/{domain_id}/inventories/{inventory_id}/modifications"
+        t0 = time.perf_counter()
+        r = self.client.post(url, json=body, timeout=120.0)
+        if r.status_code != 200:
+            raise RuntimeError(f"POST {url} -> {r.status_code}: {r.text}")
+        self.poll(domain_id, "inventories", inventory_id, timeout)
+        return time.perf_counter() - t0
+
+
+def summarize(durations: list[float]) -> dict:
+    return {
+        "n": len(durations),
+        "min": min(durations),
+        "median": statistics.median(durations),
+        "mean": statistics.fmean(durations),
+        "max": max(durations),
+    }
+
+
+def main() -> None:
+    p = argparse.ArgumentParser(description=__doc__)
+    p.add_argument("--label", required=True, help="run label, e.g. 'before'/'after'")
+    p.add_argument("--base-url", default=PROD_BASE_URL)
+    p.add_argument(
+        "--reps",
+        type=int,
+        default=5,
+        help="reps per scenario (rep 1 is a discarded warmup)",
+    )
+    p.add_argument(
+        "--half-km",
+        type=float,
+        default=1.75,
+        help="half-edge of the square domain in km (1.75 -> ~12 km^2)",
+    )
+    p.add_argument(
+        "--scoped-frac",
+        type=float,
+        default=0.25,
+        help="southern fraction of the domain the scoped mod covers",
+    )
+    p.add_argument(
+        "--factor", type=float, default=1.001, help="height multiply factor per rep"
+    )
+    p.add_argument("--seed", type=int, default=42)
+    p.add_argument(
+        "--gen-timeout",
+        type=float,
+        default=290.0,
+        help="per-resource completion timeout (s)",
+    )
+    p.add_argument("--mod-timeout", type=float, default=290.0)
+    p.add_argument("--poll-interval", type=float, default=0.3)
+    p.add_argument(
+        "--reuse-inventory", help="skip setup; benchmark this existing inventory"
+    )
+    p.add_argument("--reuse-domain", help="domain id for --reuse-inventory")
+    p.add_argument(
+        "--keep", action="store_true", help="do not delete created resources"
+    )
+    p.add_argument("--out-dir", default=str(Path(__file__).resolve().parent))
+    args = p.parse_args()
+
+    env = dotenv_values(REPO_ROOT / ".env")
+    api_key = (env.get("TEST_API_KEY") or "").strip().strip('"')
+    if not api_key:
+        raise SystemExit("TEST_API_KEY not found in repo-root .env")
+
+    bbox = square_bbox(args.half_km)
+    suffix = uuid.uuid4().hex[:8]
+    log(f"Base URL: {args.base_url}")
+    log(
+        f"Domain: {2 * args.half_km:.2f} x {2 * args.half_km:.2f} km square, bbox={bbox}"
+    )
+
+    created = {"domain": None, "grid": None, "inventory": None}
+    results = {
+        "label": args.label,
+        "base_url": args.base_url,
+        "timestamp": time.strftime("%Y-%m-%dT%H:%M:%S"),
+        "params": {
+            "reps": args.reps,
+            "half_km": args.half_km,
+            "scoped_frac": args.scoped_frac,
+            "factor": args.factor,
+            "seed": args.seed,
+        },
+    }
+
+    with Client(
+        base_url=args.base_url, headers={"API-KEY": api_key}, timeout=60.0
+    ) as client:
+        bench = Bench(client, args.poll_interval)
+        try:
+            if args.reuse_inventory:
+                domain_id = args.reuse_domain
+                inventory_id = args.reuse_inventory
+                if not domain_id:
+                    raise SystemExit("--reuse-inventory requires --reuse-domain")
+                log(f"Reusing inventory {inventory_id} in domain {domain_id}")
+            else:
+                log("Creating domain...")
+                domain_id = bench.create_domain(
+                    polygon_feature_collection(bbox, f"bench-inplace-{suffix}")
+                )
+                created["domain"] = domain_id
+                log(f"  domain {domain_id}")
+
+                log("Creating TreeMap PIM grid...")
+                grid = bench.create_and_wait(
+                    domain_id, "grids", "/grids/pim/treemap", {}, args.gen_timeout
+                )
+                created["grid"] = grid["id"]
+                log(f"  grid {grid['id']} completed")
+
+                log("Creating PIM tree inventory (point-process expansion)...")
+                inv = bench.create_and_wait(
+                    domain_id,
+                    "inventories",
+                    "/inventories/tree/pim",
+                    {"source_pim_grid_id": grid["id"], "seed": args.seed},
+                    args.gen_timeout,
+                )
+                created["inventory"] = inv["id"]
+                inventory_id = inv["id"]
+                log(f"  inventory {inventory_id} completed")
+
+            # Characterize the inventory.
+            meta = client.get(
+                f"/domains/{domain_id}/inventories/{inventory_id}/data/metadata"
+            )
+            meta.raise_for_status()
+            md = meta.json()
+            results["inventory_id"] = inventory_id
+            results["domain_id"] = domain_id
+            results["num_partitions"] = md.get("num_partitions")
+            results["total_rows"] = md.get("total_rows")
+            log(
+                f"Inventory: {md.get('total_rows')} trees across "
+                f"{md.get('num_partitions')} partitions"
+            )
+
+            scen = scenarios(bbox, args.scoped_frac, args.factor)
+            results["scenarios"] = {}
+            for name, body in scen.items():
+                log(f"Scenario '{name}': {args.reps} reps (rep 1 discarded as warmup)")
+                reps = []
+                for i in range(args.reps):
+                    dur = bench.time_modification(
+                        domain_id, inventory_id, body, args.mod_timeout
+                    )
+                    tag = "warmup" if i == 0 else "warm"
+                    log(f"  rep {i + 1}/{args.reps} ({tag}): {dur:.2f}s")
+                    reps.append(dur)
+                warm = reps[1:]
+                results["scenarios"][name] = {
+                    "cold_rep": reps[0],
+                    "warm_reps": warm,
+                    "warm_stats": summarize(warm) if warm else None,
+                }
+        finally:
+            if not args.keep and not args.reuse_inventory:
+                log("Cleaning up...")
+                for kind, rid in (
+                    ("inventories", created["inventory"]),
+                    ("grids", created["grid"]),
+                ):
+                    if rid and created["domain"]:
+                        try:
+                            client.delete(
+                                f"/domains/{created['domain']}/{kind}/{rid}",
+                                timeout=30.0,
+                            )
+                        except Exception as e:  # noqa: BLE001
+                            log(f"  warning: failed to delete {kind}/{rid}: {e}")
+                if created["domain"]:
+                    try:
+                        client.delete(
+                            f"/domains/{created['domain']}",
+                            params={"force": True},
+                            timeout=30.0,
+                        )
+                    except Exception as e:  # noqa: BLE001
+                        log(f"  warning: failed to delete domain: {e}")
+            elif created["inventory"]:
+                log(
+                    f"--keep: leaving inventory {created['inventory']} in domain {created['domain']}"
+                )
+
+    out_path = Path(args.out_dir) / f"bench_inplace_{args.label}.json"
+    out_path.write_text(json.dumps(results, indent=2) + "\n")
+    log(f"Wrote {out_path}")
+
+    print("\n=== Summary ===")
+    print(
+        f"label={results['label']}  partitions={results.get('num_partitions')}  rows={results.get('total_rows')}"
+    )
+    for name, s in results.get("scenarios", {}).items():
+        st = s.get("warm_stats")
+        if st:
+            print(
+                f"  {name:8} warm median={st['median']:.2f}s  "
+                f"mean={st['mean']:.2f}s  min={st['min']:.2f}s  "
+                f"max={st['max']:.2f}s  (cold={s['cold_rep']:.2f}s, n={st['n']})"
+            )
+
+
+if __name__ == "__main__":
+    main()
diff --git a/services/api/tests/resources/inventories/test_data_cache.py b/services/api/tests/resources/inventories/test_data_cache.py
index ac9c6a8b..d36e722a 100644
--- a/services/api/tests/resources/inventories/test_data_cache.py
+++ b/services/api/tests/resources/inventories/test_data_cache.py
@@ -1,15 +1,19 @@
 """
 Unit tests for api/resources/inventories/cache.py
 
-Tests Parquet ``_metadata`` parsing in isolation using local dask-written
-datasets. No GCP I/O.
+Inventory metadata comes from the dask DataFrame (footer-level counts), not the
+aggregated ``_metadata`` file — so it stays correct after an in-place overwrite
+that leaves ``_metadata`` stale. Exercised against local dask datasets.
 """
 
+import glob
+from unittest.mock import patch
+
 import dask.dataframe as dd
 import pandas as pd
-import pyarrow.parquet as pq
 import pytest
-from api.resources.inventories.cache import _parse_metadata
+from api.resources.inventories import cache
+from api.resources.inventories.cache import _read_metadata_sync, _read_partition_sync
 
 
 @pytest.fixture
@@ -23,28 +27,60 @@ def inventory_df() -> pd.DataFrame:
     )
 
 
-def _write_dask_parquet(df: pd.DataFrame, path: str) -> pq.FileMetaData:
-    """Write a partitioned dataset the way standgen/uploader do (#335)."""
-    dd.from_pandas(df, npartitions=2).to_parquet(
+def _write(df: pd.DataFrame, path: str, npartitions: int = 2) -> None:
+    dd.from_pandas(df, npartitions=npartitions).to_parquet(
         path, write_metadata_file=True, write_index=False
     )
-    return pq.read_metadata(f"{path}/_metadata")
 
 
-class TestParseMetadata:
-    def test_columns_match_file_schema(self, inventory_df, tmp_path):
-        metadata = _write_dask_parquet(inventory_df, str(tmp_path / "inv"))
+class TestReadMetadata:
+    def test_counts_columns_and_partitions(self, inventory_df, tmp_path):
+        path = str(tmp_path / "inv")
+        _write(inventory_df, path)
 
-        meta = _parse_metadata(metadata)
+        with patch.object(cache, "_inventory_path", return_value=path):
+            meta = _read_metadata_sync("inv")
 
+        assert meta.num_partitions == 2
+        assert meta.total_rows == 4
         assert meta.columns == ["x", "y", "height"]
 
-    def test_partitions_and_row_counts(self, inventory_df, tmp_path):
-        metadata = _write_dask_parquet(inventory_df, str(tmp_path / "inv"))
+    def test_counts_correct_after_in_place_overwrite(self, inventory_df, tmp_path):
+        """An in-place overwrite of one partition with fewer rows leaves the
+        aggregated ``_metadata`` stale; ``total_rows`` must reflect the files on
+        disk, not the stale aggregate."""
+        path = str(tmp_path / "inv")
+        _write(inventory_df, path)  # 4 rows across 2 partitions of 2
 
-        meta = _parse_metadata(metadata)
+        # Overwrite the first partition in place with a single row; do NOT touch
+        # _metadata (which still claims this partition has 2 rows / 4 total).
+        p0 = sorted(glob.glob(f"{path}/part.*.parquet"))[0]
+        pd.read_parquet(p0).iloc[:1].to_parquet(p0, index=False)
 
-        assert meta.num_partitions == 2
-        assert meta.total_rows == 4
-        assert [p.index for p in meta.partitions] == [0, 1]
-        assert sum(p.num_rows for p in meta.partitions) == 4
+        with patch.object(cache, "_inventory_path", return_value=path):
+            meta = _read_metadata_sync("inv")
+
+        assert meta.total_rows == 3  # 1 + 2, not the stale 4
+
+
+class TestReadPartition:
+    def test_reads_partition_by_index(self, inventory_df, tmp_path):
+        path = str(tmp_path / "inv")
+        _write(inventory_df, path)
+
+        with patch.object(cache, "_inventory_path", return_value=path):
+            p0 = _read_partition_sync("inv", 0, None)
+            p1 = _read_partition_sync("inv", 1, None)
+
+        assert len(p0) == 2
+        assert len(p1) == 2
+        assert list(p0.columns) == ["x", "y", "height"]
+
+    def test_reads_column_subset(self, inventory_df, tmp_path):
+        path = str(tmp_path / "inv")
+        _write(inventory_df, path)
+
+        with patch.object(cache, "_inventory_path", return_value=path):
+            p0 = _read_partition_sync("inv", 0, ["x"])
+
+        assert list(p0.columns) == ["x"]
diff --git a/services/api/tests/resources/inventories/test_data_router.py b/services/api/tests/resources/inventories/test_data_router.py
index b723b369..da7954a8 100644
--- a/services/api/tests/resources/inventories/test_data_router.py
+++ b/services/api/tests/resources/inventories/test_data_router.py
@@ -127,11 +127,7 @@ def test_returns_200(
         assert data["total_rows"] > 0
         assert isinstance(data["columns"], list)
         assert len(data["columns"]) > 0
-        assert len(data["partitions"]) == data["num_partitions"]
-        for p in data["partitions"]:
-            assert "index" in p
-            assert "num_rows" in p
-            assert p["num_rows"] > 0
+        assert "partitions" not in data
 
     def test_inventory_not_completed_returns_422(
         self, client, domain_for_testing, pending_inventory_in_firestore
diff --git a/services/api/tests/resources/inventories/test_schema.py b/services/api/tests/resources/inventories/test_schema.py
index 69573340..7f56213b 100644
--- a/services/api/tests/resources/inventories/test_schema.py
+++ b/services/api/tests/resources/inventories/test_schema.py
@@ -14,7 +14,6 @@
     InventoryDataMetadata,
     InventoryDataResponse,
     InventoryJsonOrientation,
-    InventoryPartitionInfo,
     InventorySortField,
     InventoryType,
     ListInventoriesResponse,
@@ -335,18 +334,6 @@ def test_from_string(self):
         assert InventoryJsonOrientation("records") == InventoryJsonOrientation.records
 
 
-class TestInventoryPartitionInfo:
-    def test_basic(self):
-        p = InventoryPartitionInfo(index=0, num_rows=1000)
-        assert p.index == 0
-        assert p.num_rows == 1000
-
-    def test_serialization(self):
-        p = InventoryPartitionInfo(index=2, num_rows=500)
-        d = p.model_dump()
-        assert d == {"index": 2, "num_rows": 500}
-
-
 class TestInventoryDataMetadata:
     def test_basic(self):
         meta = InventoryDataMetadata(
@@ -354,16 +341,11 @@ def test_basic(self):
             num_partitions=2,
             total_rows=2000,
             columns=["x", "y", "dbh"],
-            partitions=[
-                InventoryPartitionInfo(index=0, num_rows=1000),
-                InventoryPartitionInfo(index=1, num_rows=1000),
-            ],
         )
         assert meta.inventory_id == "abc123"
         assert meta.num_partitions == 2
         assert meta.total_rows == 2000
         assert meta.columns == ["x", "y", "dbh"]
-        assert len(meta.partitions) == 2
 
     def test_serialization_round_trip(self):
         meta = InventoryDataMetadata(
@@ -371,7 +353,6 @@ def test_serialization_round_trip(self):
             num_partitions=1,
             total_rows=100,
             columns=["x"],
-            partitions=[InventoryPartitionInfo(index=0, num_rows=100)],
         )
         d = meta.model_dump()
         restored = InventoryDataMetadata(**d)
diff --git a/services/standgen/standgen/handlers/modifications.py b/services/standgen/standgen/handlers/modifications.py
index 525eb4d9..8aa9d15b 100644
--- a/services/standgen/standgen/handlers/modifications.py
+++ b/services/standgen/standgen/handlers/modifications.py
@@ -15,7 +15,7 @@
     apply_modifications,
     resolve_spatial_conditions,
 )
-from standgen.storage import load_inventory_parquet, save_parquet_replace
+from standgen.storage import write_changed_partitions
 
 logger = logging.getLogger(__name__)
 
@@ -43,22 +43,21 @@ def apply_in_place_modifications(
     inventory_id = inventory["id"]
     modifications = inventory.get("pending_modifications", [])
 
-    # Load the inventory's own current data as a dask DataFrame.
-    progress("Loading inventory...", 10)
-    ddf = load_inventory_parquet(inventory_id)
-
-    # Apply only the new delta. Resolve spatial-condition geometries once here
-    # (off the per-partition path) when any are present.
+    # Resolve spatial-condition geometries once here (off the per-partition path)
+    # when any are present.
     progress("Applying modifications...", 40)
     if _has_spatial_condition(modifications):
         modifications = resolve_spatial_conditions(
             modifications, inventory["domain_id"], domain_gdf.crs
         )
-    ddf = ddf.map_partitions(apply_modifications, modifications)
 
-    # Replace the inventory's Parquet in place (staging swap — see storage.py).
+    # Apply the pending delta to each partition and rewrite only the partitions
+    # whose content changed — a scoped modification touches only the partitions
+    # it overlaps. The ``modifications`` ledger and ``georeference`` are unchanged.
     progress("Writing modified inventory...", 70)
-    save_parquet_replace(inventory_id, ddf)
+    write_changed_partitions(
+        inventory_id, lambda df: apply_modifications(df, modifications)
+    )
 
     progress("Complete", 100)
 
diff --git a/services/standgen/standgen/handlers/treatments.py b/services/standgen/standgen/handlers/treatments.py
index 8154aaf7..34ae1147 100644
--- a/services/standgen/standgen/handlers/treatments.py
+++ b/services/standgen/standgen/handlers/treatments.py
@@ -15,8 +15,18 @@
     _has_spatial_condition,
     resolve_spatial_conditions,
 )
-from standgen.storage import load_inventory_parquet, save_parquet_replace
-from standgen.treatments import DIA_COLUMN, apply_treatments
+from standgen.storage import (
+    load_inventory_parquet,
+    write_changed_partitions,
+    write_full_partitions,
+)
+from standgen.treatments import (
+    DIA_COLUMN,
+    apply_treatments,
+    apply_treatments_to_partition,
+    has_proportional_basal_area,
+    precompute_cutoffs,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -46,14 +56,12 @@ def apply_in_place_treatments(
     # reproducible (diameter/directional thinning consumes no RNG).
     seed = inventory.get("source", {}).get("seed")
 
-    # Load the inventory's own current data as a dask DataFrame.
-    progress("Loading inventory...", 10)
+    # Read the inventory's schema. Treatments thin against tree diameter; the API
+    # rejects dbh-less inventories at request time from the document's column
+    # metadata, but this checks the actual Parquet schema (footer only, no data
+    # scan) so a stale document fails with an actionable error instead of a
+    # KeyError mid-write.
     ddf = load_inventory_parquet(inventory_id)
-
-    # Treatments thin against tree diameter. The API rejects dbh-less
-    # inventories at request time from the document's column metadata; this
-    # checks the actual Parquet schema (no compute) so a stale document fails
-    # with an actionable error instead of a KeyError mid-write.
     if DIA_COLUMN not in ddf.columns:
         raise ProcessingError(
             code="TREATMENTS_REQUIRE_DBH",
@@ -68,18 +76,33 @@ def apply_in_place_treatments(
             ),
         )
 
-    # Apply only the new delta. Resolve spatial-condition geometries once here
-    # (off the per-partition path) when any are present.
+    # Resolve spatial-condition geometries once here (off the per-partition path)
+    # when any are present.
     progress("Applying treatments...", 40)
     if _has_spatial_condition(treatments):
         treatments = resolve_spatial_conditions(
             treatments, inventory["domain_id"], domain_gdf.crs
         )
-    ddf = apply_treatments(ddf, treatments, domain_gdf, seed=seed)
 
-    # Replace the inventory's Parquet in place (staging swap — see storage.py).
     progress("Writing treated inventory...", 70)
-    save_parquet_replace(inventory_id, ddf)
+    if has_proportional_basal_area(treatments):
+        # Proportional basal-area removes trees at random over the whole treated
+        # population, so it cannot be reproduced as a per-partition filter; the
+        # result is materialized and the full dataset rewritten.
+        result = apply_treatments(ddf, treatments, domain_gdf, seed=seed).compute()
+        write_full_partitions(inventory_id, result)
+    else:
+        # Diameter and directional basal-area thins are per-partition filters —
+        # the latter via a diameter cutoff precomputed over the treated
+        # population — so rewrite only the partitions that change, including for
+        # polygon-scoped thins.
+        cutoffs = precompute_cutoffs(ddf, treatments, domain_gdf)
+        write_changed_partitions(
+            inventory_id,
+            lambda df: apply_treatments_to_partition(
+                df, treatments, cutoffs, domain_gdf
+            ),
+        )
 
     progress("Complete", 100)
 
diff --git a/services/standgen/standgen/storage.py b/services/standgen/standgen/storage.py
index 1ce3339e..786fbf87 100644
--- a/services/standgen/standgen/storage.py
+++ b/services/standgen/standgen/storage.py
@@ -1,5 +1,6 @@
 """Storage utilities for Standgen."""
 
+import io
 import logging
 
 import dask.dataframe as dd
@@ -8,8 +9,7 @@
 import xarray as xr
 
 from lib.config import GRIDS_BUCKET, INVENTORIES_BUCKET, TABLES_BUCKET
-from lib.errors import ProcessingError
-from lib.gcs import delete_directory, exists, get_gcsfs_client
+from lib.gcs import delete_directory, get_gcsfs_client
 from lib.zarr_utils import load_zarr
 
 logger = logging.getLogger(__name__)
@@ -57,63 +57,95 @@ def save_parquet(inventory_id: str, ddf: dd.DataFrame) -> str:
     return path
 
 
-def save_parquet_replace(inventory_id: str, ddf: dd.DataFrame) -> str:
-    """Replace an inventory's Parquet data in place with ``ddf``.
+def write_changed_partitions(inventory_id: str, transform) -> str:
+    """Apply ``transform`` to each Parquet partition; overwrite only those that change.
 
-    ``ddf`` is typically derived by reading the inventory's *own* current
-    Parquet (an in-place modification). Writing straight back to the live path
-    would corrupt the data: ``dd.read_parquet`` is lazy, so the read executes
-    during this write and would race the overwrite. Instead write to a staging
-    prefix, then swap it into place once the write (and its read) have
-    completed:
+    ``transform`` is a per-partition ``DataFrame -> DataFrame`` callable (e.g.
+    ``lambda df: apply_modifications(df, mods)``). Every partition is read in one
+    concurrent ``cat`` batch; ``transform`` is applied to each; only the partitions
+    whose content actually changes are written back, in one concurrent ``pipe``
+    batch, under their existing file names. A modification scoped to part of the
+    domain therefore rewrites only the partitions it overlaps.
 
-    1. write ``ddf`` to ``{id}__rev`` (reads the live ``{id}`` here),
-    2. delete the live ``{id}`` directory,
-    3. server-side copy ``{id}__rev`` -> ``{id}`` (same bucket, no egress),
-    4. delete the staging ``{id}__rev`` directory.
+    The transform is per-partition and must not change the schema (modifications
+    only transform values or drop rows). ``_metadata`` is intentionally left
+    untouched: both the API (``pd.read_parquet`` per file) and dask
+    (``dd.read_parquet`` re-reads each file footer) read the part files directly,
+    so a stale ``_metadata`` never corrupts a read — it only affects the row
+    counts the ``/data`` metadata endpoint reports after a row-removing mod.
     """
-    live_rel = f"{INVENTORIES_BUCKET}/{inventory_id}"
-    staging_rel = f"{INVENTORIES_BUCKET}/{inventory_id}__rev"
-    staging_uri = f"gs://{staging_rel}"
-
-    # Clear any staging dir left behind by a previously failed attempt.
-    if exists(staging_uri):
-        delete_directory(staging_uri)
-
-    _write_parquet(ddf, staging_uri)
-
     fs = get_gcsfs_client()
-
-    # dask wrote staging through its own gcsfs instance, and the exists() probe
-    # above cached this client's listing of the (then-absent) staging dir. Drop
-    # the stale cache so the copy and the count check below list staging fresh.
-    fs.invalidate_cache()
-    staged_files = fs.find(staging_rel)
-
-    delete_directory(f"gs://{live_rel}")
-    # Recursive copy keeps fsspec's default on_error="ignore". gcsfs's
-    # expand_path includes a synthetic directory-marker entry for the staging
-    # prefix that 404s on rewrite; "ignore" skips it (only FileNotFoundError is
-    # swallowed, so transient errors still raise). That same leniency would also
-    # skip a genuinely missing part file, so verify completeness by file count
-    # before deleting staging — never mark an incomplete dataset completed.
-    fs.copy(staging_rel, live_rel, recursive=True)
+    base = f"{INVENTORIES_BUCKET}/{inventory_id}"
+    n = _apply_changed_partitions(fs, base, transform)
+    logger.info(f"Rewrote {n} changed partition(s) for inventory {inventory_id}")
+    return f"gs://{base}"
+
+
+def _apply_changed_partitions(fs, base: str, transform) -> int:
+    """Core of :func:`write_changed_partitions`, parameterized on the filesystem
+    so it can be tested against local files. Returns the number of partitions
+    rewritten."""
+    part_paths = sorted(p for p in fs.find(base) if p.endswith(".parquet"))
+    blobs = fs.cat(part_paths)  # one concurrent read of all partitions
+
+    changed: dict[str, bytes] = {}
+    for path in part_paths:
+        old_df = pd.read_parquet(io.BytesIO(blobs[path]))
+        # Copy so an in-place transform (apply_action mutates) leaves old_df
+        # pristine for the change comparison.
+        new_df = transform(old_df.copy())
+        if not new_df.equals(old_df):
+            buf = io.BytesIO()
+            new_df.to_parquet(buf, index=False)
+            changed[path] = buf.getvalue()
+
+    if changed:
+        fs.pipe(changed)  # one concurrent write of the changed partitions
+    return len(changed)
+
+
+def write_full_partitions(
+    inventory_id: str, df: pd.DataFrame, rows_per_partition: int = 100_000
+) -> str:
+    """Replace an inventory's Parquet with a materialized DataFrame.
+
+    Used for global (basal-area) treatments, whose result is a whole-stand
+    reduction that re-partitions the data — so a per-partition diff does not
+    apply. ``df`` must already be in memory (the caller computed it), so writing
+    back over the live path does not race a lazy read. Because the partition set
+    changes, the aggregated ``_metadata`` is removed (its file list would be
+    wrong and dask trusts it for the file set) along with any stale part files
+    from a larger previous partitioning; readers then list the directory fresh.
+    """
+    fs = get_gcsfs_client()
+    base = f"{INVENTORIES_BUCKET}/{inventory_id}"
+    n = _write_full(fs, base, df, rows_per_partition)
+    logger.info(f"Wrote {n} partition(s) for inventory {inventory_id}")
+    return f"gs://{base}"
+
+
+def _write_full(fs, base: str, df: pd.DataFrame, rows_per_partition: int) -> int:
+    """Core of :func:`write_full_partitions`, parameterized on the filesystem so
+    it can be tested against local files. Returns the partition count written."""
+    n = max(1, -(-len(df) // rows_per_partition))  # ceil division
+    new: dict[str, bytes] = {}
+    for k in range(n):
+        chunk = df.iloc[k * rows_per_partition : (k + 1) * rows_per_partition]
+        buf = io.BytesIO()
+        chunk.to_parquet(buf, index=False)
+        new[f"{base}/part.{k}.parquet"] = buf.getvalue()
+
+    fs.pipe(new)  # one concurrent write of all partitions
     fs.invalidate_cache()
-    live_files = fs.find(live_rel)
-    if len(live_files) != len(staged_files):
-        raise ProcessingError(
-            code="INVENTORY_REWRITE_INCOMPLETE",
-            message=(
-                f"In-place rewrite of inventory {inventory_id} copied "
-                f"{len(live_files)} of {len(staged_files)} files; staging "
-                f"({staging_rel}) is preserved for recovery."
-            ),
-            suggestion="Retry the modification.",
-        )
-    delete_directory(staging_uri)
-
-    logger.info(f"Replaced inventory data at gs://{live_rel}")
-    return f"gs://{live_rel}"
+    obsolete = [
+        p
+        for p in fs.find(base)
+        if (p.endswith(".parquet") and p not in new)
+        or p.rsplit("/", 1)[-1] in ("_metadata", "_common_metadata")
+    ]
+    if obsolete:
+        fs.rm(obsolete)
+    return n
 
 
 def count_inventory_rows(inventory_id: str) -> int | None:
diff --git a/services/standgen/standgen/treatments.py b/services/standgen/standgen/treatments.py
index 6ee9d814..d22fc6a9 100644
--- a/services/standgen/standgen/treatments.py
+++ b/services/standgen/standgen/treatments.py
@@ -240,6 +240,135 @@ def apply_treatments(
     return ddf
 
 
+def has_proportional_basal_area(treatments: list[dict]) -> bool:
+    """True if any treatment is a proportional basal-area thin.
+
+    Proportional thinning removes trees at random over the whole treated
+    population, so — unlike diameter and directional basal-area thins — it cannot
+    be reproduced as a per-partition filter and forces a full rewrite.
+    """
+    return any(
+        t["metric"] == "basal_area" and t["method"] == "proportional"
+        for t in treatments
+    )
+
+
+def precompute_cutoffs(
+    ddf: dd.DataFrame, treatments: list[dict], domain_gdf: gpd.GeoDataFrame
+) -> list[float | None]:
+    """Diameter cutoff for each directional basal-area treatment (None otherwise).
+
+    A directional basal-area thin to a target stand basal area is equivalent to a
+    single diameter cutoff (keep the largest trees for ``from_below``, the
+    smallest for ``from_above``). Finding that cutoff needs the whole treated
+    population at once, so this materializes it once — the in-region subset for a
+    spatially-scoped thin, the whole inventory otherwise — applying any preceding
+    treatments first so a composed thin sees the right population. The cutoff is
+    then applied per-partition by :func:`apply_treatments_to_partition`, so the
+    write stays partial (only the partitions a scoped thin overlaps change).
+    """
+    cutoffs: list[float | None] = [None] * len(treatments)
+    for i, treatment in enumerate(treatments):
+        if treatment["metric"] != "basal_area":
+            continue  # diameter treatments are row-local; no global cutoff
+        conditions = treatment.get("conditions", [])
+        region_area_ha = compute_region_area_ha(domain_gdf, conditions)
+        _enforce_area_limit(region_area_ha, scoped=bool(conditions))
+        if conditions:
+            mask = ddf.map_partitions(
+                build_condition_mask, conditions, meta=pd.Series(dtype=bool)
+            )
+            pop = ddf[mask].compute().reset_index(drop=True)
+        else:
+            pop = ddf.compute().reset_index(drop=True)
+        for j in range(i):
+            pop = _apply_treatment_to_df(pop, treatments[j], cutoffs[j], domain_gdf)
+        cutoffs[i] = _directional_cutoff(pop, treatment, domain_gdf)
+    return cutoffs
+
+
+def _directional_cutoff(
+    pop: pd.DataFrame, treatment: dict, domain_gdf: gpd.GeoDataFrame
+) -> float | None:
+    """Diameter cutoff that reproduces a directional basal-area thin of ``pop``.
+
+    Runs the fastfuels-core thinner once to get the kept set, then takes the
+    boundary diameter: the smallest kept tree for ``from_below`` (keep dbh >=
+    cutoff) or the largest for ``from_above`` (keep dbh <= cutoff). Returns
+    ``None`` when the stand is already at/below target (nothing removed).
+    """
+    if pop.empty:
+        return None
+    conditions = treatment.get("conditions", [])
+    area_ha = compute_region_area_ha(domain_gdf, conditions)
+    value_per_ha = convert_treatment_value(
+        "basal_area", treatment["value"], treatment.get("unit")
+    )
+    kept = build_thinner(treatment, value_per_ha * area_ha).apply(
+        pop, dia_column_name=DIA_COLUMN
+    )
+    if len(kept) >= len(pop):
+        return None  # already at/below target — nothing removed
+    if kept.empty:
+        # Target is below even the largest tree's basal area — remove everything.
+        # The per-partition filter then drops every in-region tree.
+        return float("inf") if treatment["method"] == "from_below" else float("-inf")
+    if treatment["method"] == "from_below":
+        return float(kept[DIA_COLUMN].min())
+    return float(kept[DIA_COLUMN].max())
+
+
+def _apply_directional_cutoff(
+    df: pd.DataFrame, treatment: dict, cutoff: float | None
+) -> pd.DataFrame:
+    """Apply a precomputed directional basal-area cutoff to a single partition."""
+    if cutoff is None or df.empty:
+        return df
+    conditions = treatment.get("conditions", [])
+    in_region = (
+        build_condition_mask(df, conditions)
+        if conditions
+        else pd.Series(True, index=df.index)
+    )
+    if treatment["method"] == "from_below":
+        remove = in_region & (df[DIA_COLUMN] < cutoff)
+    else:  # from_above
+        remove = in_region & (df[DIA_COLUMN] > cutoff)
+    return df[~remove].reset_index(drop=True)
+
+
+def _apply_treatment_to_df(
+    df: pd.DataFrame,
+    treatment: dict,
+    cutoff: float | None,
+    domain_gdf: gpd.GeoDataFrame,
+) -> pd.DataFrame:
+    """Apply one per-partition-expressible treatment (diameter or directional
+    basal-area) to a DataFrame."""
+    if treatment["metric"] == "diameter":
+        return apply_single_treatment(df, treatment, domain_gdf)
+    return _apply_directional_cutoff(df, treatment, cutoff)
+
+
+def apply_treatments_to_partition(
+    df: pd.DataFrame,
+    treatments: list[dict],
+    cutoffs: list[float | None],
+    domain_gdf: gpd.GeoDataFrame,
+) -> pd.DataFrame:
+    """Apply diameter + directional basal-area treatments to a single partition.
+
+    Each diameter treatment is a row-local filter; each directional basal-area
+    treatment is applied via its precomputed cutoff (see
+    :func:`precompute_cutoffs`). Both are per-partition, so the handler rewrites
+    only the partitions that change — including for polygon-scoped thins.
+    Proportional basal-area is NOT routed here.
+    """
+    for treatment, cutoff in zip(treatments, cutoffs):
+        df = _apply_treatment_to_df(df, treatment, cutoff, domain_gdf)
+    return df
+
+
 def _apply_treatment_to_ddf(
     ddf: dd.DataFrame, treatment: dict, domain_gdf: gpd.GeoDataFrame
 ) -> dd.DataFrame:
diff --git a/services/standgen/tests/handlers/test_modifications.py b/services/standgen/tests/handlers/test_modifications.py
index 2b31b7c6..6263c289 100644
--- a/services/standgen/tests/handlers/test_modifications.py
+++ b/services/standgen/tests/handlers/test_modifications.py
@@ -1,14 +1,14 @@
 """
 Unit tests for standgen/handlers/modifications.py
 
-Tests the in-place modifications handler with mocked I/O (GCS). Verifies it
-loads the inventory's own data, applies only the pending delta, replaces the
-data in place, and carries over the existing georeference.
+The in-place modifications handler resolves spatial conditions once, then hands a
+per-partition transform to ``write_changed_partitions`` (which rewrites only the
+partitions whose content changes). These tests mock the storage call and
+exercise the captured transform against a sample partition.
 """
 
 from unittest.mock import MagicMock, patch
 
-import dask.dataframe as dd
 import geopandas as gpd
 import numpy as np
 import pandas as pd
@@ -18,11 +18,11 @@
 
 
 @pytest.fixture
-def sample_ddf():
-    """Sample dask DataFrame that load_inventory_parquet would return."""
+def sample_partition() -> pd.DataFrame:
+    """A single inventory partition, as read off Parquet."""
     rng = np.random.default_rng(42)
     n = 50
-    df = pd.DataFrame(
+    return pd.DataFrame(
         {
             "x": rng.uniform(500000, 501000, n),
             "y": rng.uniform(5200000, 5201000, n),
@@ -33,7 +33,6 @@ def sample_ddf():
             "crown_ratio": rng.uniform(0.1, 0.9, n),
         }
     )
-    return dd.from_pandas(df, npartitions=3)
 
 
 @pytest.fixture
@@ -72,63 +71,46 @@ def base_inventory():
 
 
 class TestApplyInPlaceModifications:
-    @patch("standgen.handlers.modifications.save_parquet_replace")
-    @patch("standgen.handlers.modifications.load_inventory_parquet")
-    def test_loads_own_data_replaces_in_place_and_carries_georeference(
-        self, mock_load, mock_save, base_inventory, sample_ddf, domain_gdf
+    @patch("standgen.handlers.modifications.write_changed_partitions")
+    def test_keys_off_inventory_id_and_carries_georeference(
+        self, mock_write, base_inventory, domain_gdf
     ):
-        """Loads the inventory's own data, replaces it in place, returns the
-        existing georeference unchanged."""
-        mock_load.return_value = sample_ddf
-        progress = MagicMock()
-
-        result = apply_in_place_modifications(base_inventory, domain_gdf, progress)
+        """Writes against the inventory's own ID; returns the existing
+        georeference unchanged (modifications never move the footprint)."""
+        result = apply_in_place_modifications(base_inventory, domain_gdf, MagicMock())
 
-        # Load and replace both key off the inventory's own ID (in place).
-        mock_load.assert_called_once_with("inventory-id")
-        mock_save.assert_called_once()
-        assert mock_save.call_args[0][0] == "inventory-id"
-        assert isinstance(mock_save.call_args[0][1], dd.DataFrame)
-
-        # Georeference is carried over verbatim — not recomputed.
+        mock_write.assert_called_once()
+        assert mock_write.call_args[0][0] == "inventory-id"
+        assert callable(mock_write.call_args[0][1])
         assert result["georeference"] == base_inventory["georeference"]
 
-    @patch("standgen.handlers.modifications.save_parquet_replace")
-    @patch("standgen.handlers.modifications.load_inventory_parquet")
-    def test_applies_only_the_pending_delta(
-        self, mock_load, mock_save, base_inventory, sample_ddf, domain_gdf
+    @patch("standgen.handlers.modifications.write_changed_partitions")
+    def test_transform_applies_pending_delta(
+        self, mock_write, base_inventory, domain_gdf, sample_partition
     ):
-        """Only pending_modifications is applied to the current data."""
-        mock_load.return_value = sample_ddf
-        progress = MagicMock()
+        """The captured transform applies only the pending delta (remove dbh<5)."""
+        apply_in_place_modifications(base_inventory, domain_gdf, MagicMock())
 
-        apply_in_place_modifications(base_inventory, domain_gdf, progress)
+        transform = mock_write.call_args[0][1]
+        out = transform(sample_partition.copy())
 
-        result_df = mock_save.call_args[0][1].compute()
-        # The pending delta removes dbh < 5.0.
-        assert (result_df["dbh"] >= 5.0).all()
-        assert len(result_df) < len(sample_ddf.compute())
+        assert (out["dbh"] >= 5.0).all()
+        assert len(out) < len(sample_partition)
 
-    @patch("standgen.handlers.modifications.save_parquet_replace")
-    @patch("standgen.handlers.modifications.load_inventory_parquet")
-    def test_reports_progress(
-        self, mock_load, mock_save, base_inventory, sample_ddf, domain_gdf
-    ):
-        mock_load.return_value = sample_ddf
+    @patch("standgen.handlers.modifications.write_changed_partitions")
+    def test_reports_progress(self, mock_write, base_inventory, domain_gdf):
         progress = MagicMock()
 
         apply_in_place_modifications(base_inventory, domain_gdf, progress)
 
         messages = [c[0][0] for c in progress.call_args_list]
-        assert "Loading inventory..." in messages
         assert "Applying modifications..." in messages
         assert "Writing modified inventory..." in messages
         assert "Complete" in messages
 
-    @patch("standgen.handlers.modifications.save_parquet_replace")
-    @patch("standgen.handlers.modifications.load_inventory_parquet")
+    @patch("standgen.handlers.modifications.write_changed_partitions")
     def test_multiply_delta_preserves_row_count(
-        self, mock_load, mock_save, domain_gdf, sample_ddf
+        self, mock_write, domain_gdf, sample_partition
     ):
         inventory = {
             "id": "inventory-id",
@@ -147,18 +129,16 @@ def test_multiply_delta_preserves_row_count(
                 }
             ],
         }
-        mock_load.return_value = sample_ddf
-        progress = MagicMock()
 
-        apply_in_place_modifications(inventory, domain_gdf, progress)
+        apply_in_place_modifications(inventory, domain_gdf, MagicMock())
 
-        result_df = mock_save.call_args[0][1].compute()
-        assert len(result_df) == len(sample_ddf.compute())
+        transform = mock_write.call_args[0][1]
+        out = transform(sample_partition.copy())
+        assert len(out) == len(sample_partition)
 
-    @patch("standgen.handlers.modifications.save_parquet_replace")
-    @patch("standgen.handlers.modifications.load_inventory_parquet")
+    @patch("standgen.handlers.modifications.write_changed_partitions")
     def test_multiple_pending_modifications_applied_sequentially(
-        self, mock_load, mock_save, domain_gdf, sample_ddf
+        self, mock_write, domain_gdf, sample_partition
     ):
         inventory = {
             "id": "inventory-id",
@@ -183,11 +163,10 @@ def test_multiple_pending_modifications_applied_sequentially(
                 },
             ],
         }
-        mock_load.return_value = sample_ddf
-        progress = MagicMock()
 
-        apply_in_place_modifications(inventory, domain_gdf, progress)
+        apply_in_place_modifications(inventory, domain_gdf, MagicMock())
 
-        result_df = mock_save.call_args[0][1].compute()
-        assert (result_df["dbh"] >= 3.0).all()
-        assert len(result_df) < len(sample_ddf.compute())
+        transform = mock_write.call_args[0][1]
+        out = transform(sample_partition.copy())
+        assert (out["dbh"] >= 3.0).all()
+        assert len(out) < len(sample_partition)
diff --git a/services/standgen/tests/handlers/test_treatments.py b/services/standgen/tests/handlers/test_treatments.py
index 151b9282..46e83e2b 100644
--- a/services/standgen/tests/handlers/test_treatments.py
+++ b/services/standgen/tests/handlers/test_treatments.py
@@ -1,9 +1,11 @@
 """
 Unit tests for standgen/handlers/treatments.py
 
-Tests the in-place treatments handler with mocked I/O (GCS). Verifies it loads
-the inventory's own data, applies only the pending treatment delta, replaces the
-data in place, and carries over the existing georeference.
+Diameter treatments are row-local and go through ``write_changed_partitions``
+(partial rewrite); basal-area treatments are whole-stand reductions, materialized
+and written via ``write_full_partitions``. These tests mock the storage calls and
+exercise either the captured per-partition transform (diameter) or the
+materialized result (basal-area).
 """
 
 from unittest.mock import MagicMock, patch
@@ -18,13 +20,14 @@
 
 from lib.errors import ProcessingError
 
+STORAGE = "standgen.handlers.treatments"
+
 
 @pytest.fixture
-def sample_ddf():
-    """Sample dask DataFrame that load_inventory_parquet would return."""
+def sample_partition() -> pd.DataFrame:
     rng = np.random.default_rng(42)
     n = 200
-    df = pd.DataFrame(
+    return pd.DataFrame(
         {
             "x": rng.uniform(500000, 501000, n),
             "y": rng.uniform(5200000, 5201000, n),
@@ -35,29 +38,30 @@ def sample_ddf():
             "crown_ratio": rng.uniform(0.1, 0.9, n),
         }
     )
-    return dd.from_pandas(df, npartitions=3)
+
+
+@pytest.fixture
+def sample_ddf(sample_partition):
+    return dd.from_pandas(sample_partition, npartitions=3)
 
 
 @pytest.fixture
 def domain_gdf():
-    """Domain GeoDataFrame (used for basal-area area sizing / spatial conditions)."""
     return gpd.GeoDataFrame(
-        geometry=[box(500000, 5200000, 501000, 5201000)],
-        crs="EPSG:32611",
+        geometry=[box(500000, 5200000, 501000, 5201000)], crs="EPSG:32611"
     )
 
 
 @pytest.fixture
 def small_domain_gdf():
-    """A 10 m × 10 m domain (0.01 ha). A small area makes a per-hectare basal-area
-    target a small absolute m² target, so an inventory-wide thin actually reduces
-    the synthetic stand (whose trees are sized for tests, not realistic density)."""
+    """A 10 m × 10 m domain (0.01 ha) so a per-hectare basal-area target is a
+    small absolute m² target and an inventory-wide thin actually reduces the
+    synthetic stand."""
     return gpd.GeoDataFrame(geometry=[box(0, 0, 10, 10)], crs="EPSG:32611")
 
 
 @pytest.fixture
 def base_inventory():
-    """A completed inventory with one pending diameter treatment delta to apply."""
     return {
         "id": "inventory-id",
         "domain_id": "domain-123",
@@ -73,48 +77,83 @@ def base_inventory():
     }
 
 
-class TestApplyInPlaceTreatments:
-    @patch("standgen.handlers.treatments.save_parquet_replace")
-    @patch("standgen.handlers.treatments.load_inventory_parquet")
-    def test_loads_own_data_replaces_in_place_and_carries_georeference(
-        self, mock_load, mock_save, base_inventory, sample_ddf, domain_gdf
+class TestDiameterTreatments:
+    @patch(f"{STORAGE}.write_full_partitions")
+    @patch(f"{STORAGE}.write_changed_partitions")
+    @patch(f"{STORAGE}.load_inventory_parquet")
+    def test_routes_to_partial_rewrite_and_carries_georeference(
+        self, mock_load, mock_changed, mock_full, base_inventory, sample_ddf, domain_gdf
     ):
-        """Loads the inventory's own data, replaces it in place, returns the
-        existing georeference unchanged."""
         mock_load.return_value = sample_ddf
-        progress = MagicMock()
 
-        result = apply_in_place_treatments(base_inventory, domain_gdf, progress)
+        result = apply_in_place_treatments(base_inventory, domain_gdf, MagicMock())
 
-        # Load and replace both key off the inventory's own ID (in place).
-        mock_load.assert_called_once_with("inventory-id")
-        mock_save.assert_called_once()
-        assert mock_save.call_args[0][0] == "inventory-id"
-        assert isinstance(mock_save.call_args[0][1], dd.DataFrame)
-
-        # Georeference is carried over verbatim — not recomputed.
+        mock_changed.assert_called_once()
+        assert mock_changed.call_args[0][0] == "inventory-id"
+        assert callable(mock_changed.call_args[0][1])
+        mock_full.assert_not_called()  # diameter is per-partition, not a full rewrite
         assert result["georeference"] == base_inventory["georeference"]
 
-    @patch("standgen.handlers.treatments.save_parquet_replace")
-    @patch("standgen.handlers.treatments.load_inventory_parquet")
-    def test_applies_only_the_pending_delta(
-        self, mock_load, mock_save, base_inventory, sample_ddf, domain_gdf
+    @patch(f"{STORAGE}.write_full_partitions")
+    @patch(f"{STORAGE}.write_changed_partitions")
+    @patch(f"{STORAGE}.load_inventory_parquet")
+    def test_transform_thins_below_diameter(
+        self,
+        mock_load,
+        mock_changed,
+        mock_full,
+        base_inventory,
+        sample_ddf,
+        sample_partition,
+        domain_gdf,
     ):
-        """Only pending_treatments is applied to the current data."""
         mock_load.return_value = sample_ddf
-        progress = MagicMock()
+        apply_in_place_treatments(base_inventory, domain_gdf, MagicMock())
+
+        transform = mock_changed.call_args[0][1]
+        out = transform(sample_partition.copy())
+
+        assert (out["dbh"] >= 10.0).all()
+        assert len(out) < len(sample_partition)
+
+    @patch(f"{STORAGE}.write_full_partitions")
+    @patch(f"{STORAGE}.write_changed_partitions")
+    @patch(f"{STORAGE}.load_inventory_parquet")
+    def test_multiple_diameter_treatments_compose(
+        self,
+        mock_load,
+        mock_changed,
+        mock_full,
+        sample_ddf,
+        sample_partition,
+        domain_gdf,
+    ):
+        inventory = {
+            "id": "inventory-id",
+            "domain_id": "domain-123",
+            "georeference": {"crs": "EPSG:32611", "bounds": [0, 0, 1, 1]},
+            "source": {"name": "pim", "seed": 3},
+            "treatments": [],
+            "pending_treatments": [
+                {"metric": "diameter", "method": "from_below", "value": 5.0},
+                {"metric": "diameter", "method": "from_above", "value": 45.0},
+            ],
+        }
+        mock_load.return_value = sample_ddf
+        apply_in_place_treatments(inventory, domain_gdf, MagicMock())
 
-        apply_in_place_treatments(base_inventory, domain_gdf, progress)
+        transform = mock_changed.call_args[0][1]
+        out = transform(sample_partition.copy())
 
-        result_df = mock_save.call_args[0][1].compute()
-        # The pending diameter thin-from-below removes trees below 10 cm dbh.
-        assert (result_df["dbh"] >= 10.0).all()
-        assert len(result_df) < len(sample_ddf.compute())
+        assert (out["dbh"] >= 5.0).all()
+        assert (out["dbh"] <= 45.0).all()
+        assert len(out) < len(sample_partition)
 
-    @patch("standgen.handlers.treatments.save_parquet_replace")
-    @patch("standgen.handlers.treatments.load_inventory_parquet")
+    @patch(f"{STORAGE}.write_full_partitions")
+    @patch(f"{STORAGE}.write_changed_partitions")
+    @patch(f"{STORAGE}.load_inventory_parquet")
     def test_reports_progress(
-        self, mock_load, mock_save, base_inventory, sample_ddf, domain_gdf
+        self, mock_load, mock_changed, mock_full, base_inventory, sample_ddf, domain_gdf
     ):
         mock_load.return_value = sample_ddf
         progress = MagicMock()
@@ -122,18 +161,26 @@ def test_reports_progress(
         apply_in_place_treatments(base_inventory, domain_gdf, progress)
 
         messages = [c[0][0] for c in progress.call_args_list]
-        assert "Loading inventory..." in messages
         assert "Applying treatments..." in messages
         assert "Writing treated inventory..." in messages
         assert "Complete" in messages
 
-    @patch("standgen.handlers.treatments.save_parquet_replace")
-    @patch("standgen.handlers.treatments.load_inventory_parquet")
-    def test_basal_area_proportional_reduces_stand(
-        self, mock_load, mock_save, small_domain_gdf, sample_ddf
+
+class TestDirectionalBasalArea:
+    @patch(f"{STORAGE}.write_full_partitions")
+    @patch(f"{STORAGE}.write_changed_partitions")
+    @patch(f"{STORAGE}.load_inventory_parquet")
+    def test_routes_to_partial_rewrite_via_cutoff(
+        self,
+        mock_load,
+        mock_changed,
+        mock_full,
+        domain_gdf,
+        sample_ddf,
+        sample_partition,
     ):
-        """An inventory-wide proportional basal-area thin reduces the stand. This
-        exercises the materialized basal-area path and the seed plumbing."""
+        """A directional basal-area thin is reduced to a diameter cutoff and goes
+        through the per-partition partial-rewrite path — not a full rewrite."""
         inventory = {
             "id": "inventory-id",
             "domain_id": "domain-123",
@@ -141,52 +188,64 @@ def test_basal_area_proportional_reduces_stand(
             "source": {"name": "pim", "seed": 7},
             "treatments": [],
             "pending_treatments": [
-                {"metric": "basal_area", "method": "proportional", "value": 5.0}
+                {"metric": "basal_area", "method": "from_below", "value": 0.05}
             ],
         }
         mock_load.return_value = sample_ddf
-        progress = MagicMock()
 
-        apply_in_place_treatments(inventory, small_domain_gdf, progress)
+        apply_in_place_treatments(inventory, domain_gdf, MagicMock())
+
+        mock_changed.assert_called_once()
+        mock_full.assert_not_called()  # directional is a per-partition cutoff
+        # The captured per-partition transform thins from below (removes the
+        # smallest trees), so the partition shrinks and its minimum dbh rises.
+        transform = mock_changed.call_args[0][1]
+        out = transform(sample_partition.copy())
+        assert len(out) < len(sample_partition)
+        assert out["dbh"].min() > sample_partition["dbh"].min()
 
-        result_df = mock_save.call_args[0][1].compute()
-        assert len(result_df) < len(sample_ddf.compute())
 
-    @patch("standgen.handlers.treatments.save_parquet_replace")
-    @patch("standgen.handlers.treatments.load_inventory_parquet")
-    def test_multiple_pending_treatments_applied_sequentially(
-        self, mock_load, mock_save, domain_gdf, sample_ddf
+class TestProportionalBasalArea:
+    @patch(f"{STORAGE}.write_full_partitions")
+    @patch(f"{STORAGE}.write_changed_partitions")
+    @patch(f"{STORAGE}.load_inventory_parquet")
+    def test_routes_to_full_rewrite_and_reduces_stand(
+        self, mock_load, mock_changed, mock_full, small_domain_gdf, sample_ddf
     ):
+        """A proportional basal-area thin removes trees at random, so it cannot be
+        a per-partition filter — it materializes the reduced stand and is written
+        via the full-rewrite path. Also exercises the seed plumbing."""
         inventory = {
             "id": "inventory-id",
             "domain_id": "domain-123",
             "georeference": {"crs": "EPSG:32611", "bounds": [0, 0, 1, 1]},
-            "source": {"name": "pim", "seed": 3},
+            "source": {"name": "pim", "seed": 7},
             "treatments": [],
             "pending_treatments": [
-                {"metric": "diameter", "method": "from_below", "value": 5.0},
-                {"metric": "diameter", "method": "from_above", "value": 45.0},
+                {"metric": "basal_area", "method": "proportional", "value": 5.0}
             ],
         }
         mock_load.return_value = sample_ddf
-        progress = MagicMock()
 
-        apply_in_place_treatments(inventory, domain_gdf, progress)
+        apply_in_place_treatments(inventory, small_domain_gdf, MagicMock())
 
-        result_df = mock_save.call_args[0][1].compute()
-        # Both diameter cutoffs applied: 5.0 <= dbh <= 45.0.
-        assert (result_df["dbh"] >= 5.0).all()
-        assert (result_df["dbh"] <= 45.0).all()
-        assert len(result_df) < len(sample_ddf.compute())
+        mock_full.assert_called_once()
+        assert mock_full.call_args[0][0] == "inventory-id"
+        result_df = mock_full.call_args[0][1]
+        assert isinstance(result_df, pd.DataFrame)
+        assert len(result_df) < 200  # the stand was thinned
+        mock_changed.assert_not_called()
 
-    @patch("standgen.handlers.treatments.save_parquet_replace")
-    @patch("standgen.handlers.treatments.load_inventory_parquet")
-    def test_data_without_dbh_raises_actionable_error(
-        self, mock_load, mock_save, base_inventory, domain_gdf
+
+class TestDbhValidation:
+    @patch(f"{STORAGE}.write_full_partitions")
+    @patch(f"{STORAGE}.write_changed_partitions")
+    @patch(f"{STORAGE}.load_inventory_parquet")
+    def test_data_without_dbh_raises_before_any_write(
+        self, mock_load, mock_changed, mock_full, base_inventory, domain_gdf
     ):
-        """Data with no dbh column (e.g. a CHM-derived inventory whose document
-        metadata predates the column fix) fails with an actionable
-        ProcessingError before any compute or write — not a KeyError."""
+        """Data with no dbh column fails with an actionable ProcessingError
+        before any write — not a KeyError."""
         df = pd.DataFrame(
             {
                 "x": [500000.0, 500500.0],
@@ -195,10 +254,10 @@ def test_data_without_dbh_raises_actionable_error(
             }
         )
         mock_load.return_value = dd.from_pandas(df, npartitions=1)
-        progress = MagicMock()
 
         with pytest.raises(ProcessingError) as exc_info:
-            apply_in_place_treatments(base_inventory, domain_gdf, progress)
+            apply_in_place_treatments(base_inventory, domain_gdf, MagicMock())
 
         assert exc_info.value.code == "TREATMENTS_REQUIRE_DBH"
-        mock_save.assert_not_called()
+        mock_changed.assert_not_called()
+        mock_full.assert_not_called()
diff --git a/services/standgen/tests/test_storage.py b/services/standgen/tests/test_storage.py
index addcc57f..14465823 100644
--- a/services/standgen/tests/test_storage.py
+++ b/services/standgen/tests/test_storage.py
@@ -5,11 +5,14 @@
 No GCP I/O.
 """
 
+import os
+
 import dask.dataframe as dd
+import fsspec
 import pandas as pd
 import pyarrow.parquet as pq
 import pytest
-from standgen.storage import _write_parquet
+from standgen.storage import _apply_changed_partitions, _write_full, _write_parquet
 
 
 @pytest.fixture
@@ -64,3 +67,99 @@ def test_legacy_dataset_rewrites_clean(self, inventory_df, tmp_path):
         assert _schema_names(rewritten_path) == ["x", "y", "height"]
         result = dd.read_parquet(rewritten_path).compute().reset_index(drop=True)
         pd.testing.assert_frame_equal(result, inventory_df)
+
+
+class TestApplyChangedPartitions:
+    """The partial-rewrite core: apply a per-partition transform and overwrite
+    only the partitions that change. Exercised against the local filesystem."""
+
+    @staticmethod
+    def _write_parts(base: str, dfs: list[pd.DataFrame]) -> None:
+        os.makedirs(base, exist_ok=True)
+        for k, df in enumerate(dfs):
+            df.to_parquet(f"{base}/part.{k}.parquet", index=False)
+
+    @staticmethod
+    def _raw(path: str) -> bytes:
+        with open(path, "rb") as f:
+            return f.read()
+
+    def test_writes_only_the_changed_partition(self, tmp_path):
+        """A transform scoped to one partition's rows rewrites only that
+        partition; the others stay byte-identical (never re-uploaded)."""
+        base = str(tmp_path / "inv")
+        # Three spatially-separated partitions: x in [0,5), [10,15), [20,25).
+        dfs = [
+            pd.DataFrame({"x": [i * 10 + j for j in range(5)], "dbh": [10.0] * 5})
+            for i in range(3)
+        ]
+        self._write_parts(base, dfs)
+        p1_before = self._raw(f"{base}/part.1.parquet")
+        p2_before = self._raw(f"{base}/part.2.parquet")
+
+        def transform(df):
+            df.loc[df["x"] < 5, "dbh"] *= 0.5  # only partition 0 has x < 5
+            return df
+
+        n = _apply_changed_partitions(fsspec.filesystem("local"), base, transform)
+
+        assert n == 1
+        assert pd.read_parquet(f"{base}/part.0.parquet")["dbh"].tolist() == [5.0] * 5
+        # Untouched partitions are not rewritten at all — byte-for-byte identical.
+        assert self._raw(f"{base}/part.1.parquet") == p1_before
+        assert self._raw(f"{base}/part.2.parquet") == p2_before
+
+    def test_no_change_writes_nothing(self, tmp_path):
+        base = str(tmp_path / "inv")
+        dfs = [
+            pd.DataFrame({"x": [1.0, 2, 3], "dbh": [10.0, 11, 12]}) for _ in range(2)
+        ]
+        self._write_parts(base, dfs)
+        before = {k: self._raw(f"{base}/part.{k}.parquet") for k in range(2)}
+
+        n = _apply_changed_partitions(fsspec.filesystem("local"), base, lambda df: df)
+
+        assert n == 0
+        for k in range(2):
+            assert self._raw(f"{base}/part.{k}.parquet") == before[k]
+
+    def test_remove_shrinks_only_matching_partition(self, tmp_path):
+        base = str(tmp_path / "inv")
+        dfs = [
+            pd.DataFrame({"x": [0.0, 1, 2], "dbh": [1.0, 2, 3]}),  # all dbh < 5
+            pd.DataFrame({"x": [10.0, 11, 12], "dbh": [20.0, 21, 22]}),  # all >= 5
+        ]
+        self._write_parts(base, dfs)
+
+        def transform(df):
+            return df[df["dbh"] >= 5.0].reset_index(drop=True)
+
+        n = _apply_changed_partitions(fsspec.filesystem("local"), base, transform)
+
+        assert n == 1
+        assert len(pd.read_parquet(f"{base}/part.0.parquet")) == 0
+        assert len(pd.read_parquet(f"{base}/part.1.parquet")) == 3
+
+
+class TestWriteFull:
+    """The full-rewrite path (used for proportional basal-area). Replaces the
+    dataset with a materialized DataFrame, dropping the now-wrong ``_metadata``
+    and any stale part files so dask re-lists the directory."""
+
+    def test_rewrites_and_drops_metadata_and_stale_parts(self, tmp_path):
+        base = str(tmp_path / "inv")
+        dd.from_pandas(
+            pd.DataFrame({"x": range(300), "dbh": [10.0] * 300}), npartitions=3
+        ).to_parquet(base, write_metadata_file=True, write_index=False)
+        assert "_metadata" in os.listdir(base)
+
+        new_df = pd.DataFrame({"x": range(50), "dbh": [10.0] * 50})
+        n = _write_full(
+            fsspec.filesystem("local"), base, new_df, rows_per_partition=100_000
+        )
+
+        assert n == 1
+        # Only the new partition remains: stale parts, _metadata, _common_metadata
+        # are all removed so readers list the directory fresh.
+        assert set(os.listdir(base)) == {"part.0.parquet"}
+        assert len(pd.read_parquet(base)) == 50