diff --git a/docs/smart-row-cache-redesign.md b/docs/smart-row-cache-redesign.md
index ab8aacbd5..5e5429ec3 100644
--- a/docs/smart-row-cache-redesign.md
+++ b/docs/smart-row-cache-redesign.md
@@ -1,7 +1,46 @@
 # SmartRowCache redesign — rowid-keyed store + view permutations
 
 ## Status
-Phases 1-6 + controller layer of phase 7 landed on `feat/smart-row-cache-redesign`. 183 JS tests + 11 Python tests pass. Remaining work (consumer cutover) is listed at the bottom.
+Phases 1-7a landed (PR #719, merged to `main`); the negative-start `rowidsInRange` bug found while evaluating them was fixed in #942. Design decisions for the consumer cutover were resolved 2026-06-25 — see [Decisions resolved (2026-06-25)](#decisions-resolved-2026-06-25); they supersede the original "Decisions" set where they conflict. The remaining consumer cutover (phases 7b-7e) is planned in detail at the bottom.
+
+## Decisions resolved (2026-06-25)
+
+Resolving the two blockers found while evaluating phase 1-7a. These supersede the conflicting bullets under "Decisions" below.
+
+### A — `getRowsByRowid([…])` is in v1 (reversed)
+
+After a sort, the visible window maps to an arbitrary, non-contiguous set of rowids. A positional `populate(start, end)` against the *unsorted* tagged expression cannot fetch them, and re-`order_by`-ing the whole expression per window is exactly the cost this redesign removes. The fetch-by-rowid path is the one the JS side already speaks — `missingAt(view, start, end)` returns the missing rowids, and:
+
+```
+getRowsByRowid({sourceName, rowids: Int32Array})
+  → {rowids: Int32Array, rows: Row[]}
+```
+
+fetches exactly those. `populate(start, end)` stays as the cheap path for the default (identity) view and eager head/tail prefetch; `getRowsByRowid` is the general path whenever a non-identity view is active.
+
+### B — every view is an ordered list of `original_row_id`s
+
+The rowid is the **`original_row_id`**: the stable identity of a row in the source frame as delivered, decoupled from display position. `RowStore` is keyed by it, and `SortView`/`FilterView` are already just `Int32Array`s of these ids — so they unify into one ordered-rowid-list view (`IdentityView` stays the privileged no-array case):
+
+- sort → all N ids, reordered
+- filter / search → a subset of ids
+- **filter + sort → the filtered subset in sort order** — still one ordered id list
+
+Because everything shares the one id namespace, "filtered then sorted" composes for free, client-side, no extra round-trip:
+
+```
+combined = sortView.rowidOrder.filter(r => filterSet.has(r))
+```
+
+Prefer this client-side intersection (the client already holds the full sort permutation and the filter subset); fall back to a server-composed combined list only when one input is absent. This also resolves the "scrambled filter order" issue: a filter-only view in original order is the subset ascending by `original_row_id`.
+
+### The cache key is the *generation*, not the sort
+
+`KeyAwareSmartRowCache` selects a cache by `${sourceName}-${sort}-${sort_direction}` (`getSourcePayloadKey`), conflating content identity with sort — the root reason a sort refetches everything. The new selection key is the **generation key**: only the ops that *change row content* (postprocessor, low-code mutate), classified by the lisp "changes rows" annotation. Sort and filter come *out* of the key and become views inside one `RowCache`.
+
+`Map<generationKey, RowCache>` replaces `Map<getSourcePayloadKey, SmartRowCache>`; `KeyAwareSmartRowCache`'s outer-map + callback machinery is lifted into the new controller — only the key *definition* changes. `sourceName` (= `JSON.stringify(outside_df_params)`) currently folds filter/search in, so the cutover must split `outside_df_params` along the same annotation: sort + filter/search → views, content-changers → generation key.
+
+`original_row_id` is stable only *within* a generation; a content-changing op re-tags and resets the namespace, invalidating that generation's views and `RowStore`.
 
 ## Today's architecture (briefly)
 
@@ -26,7 +65,9 @@ view V at position k  →  rowid r = V.rowidOrder[k]  →  RowStore[r]
 
 The existing `KeyAwareSmartRowCache` machinery survives, but only for views that **change row contents** (mutate, group-by, join). Sort and filter share the source RowStore.
 
-## Decisions (all locked)
+## Decisions (original set)
+
+> Superseded where noted by [Decisions resolved (2026-06-25)](#decisions-resolved-2026-06-25).
 
 ### What is a rowid
 Server-assigned monotonic int from the order of the original DataFrame as delivered to the widget. Stable across the session. Decoupled from any user-visible column (not pandas index, not xorq's `index` column — they may exist as columns in their own right, but they are not rowids).
@@ -48,8 +89,9 @@ Boundary baked into the lisp annotation:
 | postprocessing func | yes | fresh DerivedRowStore |
 | most low-code (lisp) ops | yes | fresh DerivedRowStore |
 | low-code op annotated as filter-only | no | `FilterView` |
+| auto-clean (additive columns) | no | v1: generation reset; v2: column delta — see [Forward compatibility](#forward-compatibility) |
 
-The lisp return value carries an annotation flag. Search is the canonical "filter-only" op.
+The lisp return value carries an annotation flag. Search is the canonical "filter-only" op. Auto-cleaning is row-preserving and additive at the column level, so it is a generation reset in v1 only for simplicity — see [Forward compatibility](#forward-compatibility).
 
 ### Server contract
 
@@ -80,7 +122,7 @@ filter({sourceName, filterKey})
 
 No row payload — the subset is rowids, rows are already in the rowstore (or fetched on demand).
 
-A `getRowsByRowid([…])` endpoint is **not** in v1. Adding it later is cheap if scrolling patterns demand it.
+A `getRowsByRowid([…])` endpoint **is** in v1 (reversed 2026-06-25 — see [Decisions resolved → A](#decisions-resolved-2026-06-25)). It is the general fetch path once any non-identity view is active.
 
 ### RowStore GC
 Two policies, both active:
@@ -136,17 +178,65 @@ Phases 1-5 are pure TS, no Python touched. Phase 6 is the Python contract. Phase
 | `…/RowCache.ts` | Integration controller — the API the consumer uses |
 | `buckaroo/row_cache_payloads.py` | `tag_with_rowids` + populate/sort/filter response builders |
 
-## What's left (next session)
+## Remaining implementation plan (7b-7e)
+
+Phases 1-7a shipped the primitives (`RowStore`, the unified view, GC, registry, `RowCache` controller) and the Python payload builders. None of it is wired to a consumer: `TableInfinite.tsx` is a dead stub (`<h1>broken</h1>`); the live path is `BuckarooWidgetInfinite.tsx::getKeySmartRowCache` → `getDs` (`gridUtils.ts`) → `KeyAwareSmartRowCache`, and Python dispatches `infinite_request` → `_handle_payload_args` (three near-duplicate copies: `buckaroo_widget.py`, `polars_buckaroo.py`, `xorq_buckaroo.py`) via `payload_bridge`.
+
+### API-surface gaps in 1-7a the consumer needs first
+
+`RowCache` is pure state. Before any wiring it needs:
+
+1. **Request/callback orchestration.** "fire request, park callback, replay on response, dedupe in-flight, eager head/tail prefetch" — all of which `KeyAwareSmartRowCache.getRequestRows`/`addPayloadResponse` provide today and `RowCache` does not. This is the bulk of 7b. Lift it from `KeyAwareSmartRowCache` into a `RowCacheController`.
+2. **Eager head/tail helper.** GC *pins* head/tail; the controller must compute the `[0, head)` / `[N-tail, N)` windows to fetch after a sort lands. Add `RowCache.headTailWindowsToFetch(view)`.
+3. **`getRowsByRowid` plumbing** (decision A) — JS request trigger + Python builder.
+4. **pandas/polars tagger.** `tag_with_rowids` is xorq-only (`to_pyarrow()` + memtable). The pandas/polars handlers need a parallel int32 `_buckaroo_rowid` appender before they can serve any payload.
+
+### 7b — Wire RowCache into the datasource (JS)
+
+New `gridUtils.ts::getRowCacheDs()` (parallel to `getDs`); `BuckarooWidgetInfinite.tsx` builds a `RowCacheController` (the `Map<generationKey, RowCache>` + callback bookkeeping lifted from `KeyAwareSmartRowCache`); `getRowId` switches from `data.index` to `data._buckaroo_rowid`. The datasource translates AG-Grid's positional `getRows(start, end, sortModel)` into:
+
+1. Resolve the active view — no sortModel → `defaultView()`; sortModel → `getSortView(col, dir)`, and if absent, fire a `sort` request and park the callback until `rowidOrder` lands (the view doesn't exist until the permutation arrives — the central new async step). Compose an active filter by client-side intersection (decision B).
+2. `missingAt(view, start, end)` → if empty, `successCallback(rowsAt(...), view.length())` + `gc(...)`. If non-empty, fetch the missing rowids via `getRowsByRowid` (or positional `populate` for the identity view), park the callback, replay on response.
+
+**Biggest risk:** sort is now two round-trips before a visible row (sort → `rowidOrder` → fetch rows), where today it's one. AG-Grid has no "permutation loading" state. The controller must guarantee exactly one callback resolution per `getRows` across the chain and dedupe an in-flight `sort` for the same view, or it deadlocks / double-fetches.
+
+### 7c — Wire row_cache_payloads into the Python handler
+
+Extend `payload_bridge` to route `msg.type ∈ {populate, sort, filter, getRowsByRowid}` to new handlers, leaving the legacy `infinite_request` path intact during cutover. `tag_with_rowids` runs **once at widget init** (rowids stable for the session) and re-tags only on a generation change. Build the pandas/polars tagger first (TDD, mirror `test_row_cache_payloads.py`).
+
+**Total-order invariant (most important correctness rule):** append `_buckaroo_rowid` as the final `order_by` tiebreaker in *both* `make_sort_payload` and any populate-under-sort path. `order_by` on a non-unique column has nondeterministic tie order across executions; if the sort's order and a fetch's order disagree, the grid shows wrong rows with no error. Decision A (fetch by rowid, not by position-in-sort) removes most of this exposure, but keep the tiebreaker.
+
+### 7d — Playwright / fixtures
+
+The Storybook story fakes (`SmallDFScroll`, `BuckarooWidgetTest`, `FitContentHeight`) instantiate `KeyAwareSmartRowCache` + `getDs` directly and reply via `setTimeout`; rewrite them to speak the new message kinds. Keep one story on the old path while `KeyAwareSmartRowCache` survives. The two-round-trip sort means any "click sort → assert content immediately" test must wait on the row fetch, not just the sort.
+
+### 7e — Delete dead code
+
+`KeyAwareSmartRowCache`'s per-generation map + callback machinery survives (lifted into the controller); only `SmartRowCache`'s segment algebra and `SmartRowCache.test.ts` retire, plus the `TableInfinite.tsx` stub. Do it as a pure-deletion commit after 7d is green; keep the legacy `infinite_request` path until then for A/B at any commit. Keep `PayloadArgs`/`PayloadResponse`/`getPayloadKey` until every importer is moved.
+
+## Forward compatibility
+
+The redesign decouples row *identity* (`original_row_id`) from row *display position*. The same move applies later to the column axis; v1 must not foreclose it.
+
+**Auto-cleaning is a deliberate v1 generation reset, but is row-preserving and additive.** Changing `dates_as_strings` yields a parsed `dates_as_strings` plus a renamed `dates_as_strings_original` — the original values are untouched, just relabeled, and the parsed values arrive as a new column. So a future optimization can treat cleaning as a **column delta on the same generation** rather than a reset:
+
+- additive clean → keep the `RowStore`, fetch only the new columns (a column-projected `populate`)
+- a future row-dropping clean ("remove these outliers") → a `FilterView` (subset) over the same generation
+
+For v1, cleaning stays a generation reset — simple, correct, and cleaning-at-load has no warm cache to preserve. The payoff of the column-delta path is the `cleaning_method` *toggle* on an already-scrolled grid.
+
+**The v2 hook is `original_col_id`** — the column-axis mirror of `original_row_id`: a stable per-column id assigned at first tag and carried through cleaning (the renamed original keeps its id; the new column gets a fresh one). The cell cache then keys on `(original_row_id, original_col_id)`, and the internal a,b,c names, display names, and the `_original` suffix become pure presentation. The internal a,b,c is **positional** today — the column-axis analog of display position — so it must never become a persistent cache key.
+
+**Two guardrails keep the door open at ~zero v1 cost:**
+
+1. Route every op through the single "changes rows / reorder / subset / changes columns" classifier — even though cleaning maps to "reset" in v1 — so it can be reclassified to "column delta" later in one place.
+2. Never bake the positional a,b,c (or any display position) into a persistent cache key, on either axis.
 
-- Wire `RowCache` into `TableInfinite.tsx` so the AG-Grid datasource pulls from it instead of `KeyAwareSmartRowCache`.
-- Wire `row_cache_payloads` into the Python widget message handler (new `populate` / `sort` / `filter` message kinds).
-- Decide whether to keep `KeyAwareSmartRowCache` (design says yes, for the "postprocessor-changed-rows" path) or fully replace.
-- Verify Playwright tests pass against the new wire format; update fixtures.
-- Delete dead code: at minimum, `SmartRowCache.ts` and `SmartRowCache.test.ts` once `KeyAwareSmartRowCache` no longer depends on them.
+**The one v1 simplification v2 revisits:** `RowStore` is row-granular — `has(rowid)` / `missingAt` assume "rowid present ⇒ all columns present". A column-delta clean breaks that. Refining to `(rowid, colid)` granularity is additive and localized (`RowStore.has`/`missingAt` + the datasource fetch loop), not a teardown.
 
 ## Out of scope for v1
 
-- `getRowsByRowid([…])` — only if scroll patterns prove it necessary.
 - Datasets >10M rows — hard cap on whole-dataset permutations. If we want to go bigger, we revisit windowed permutations.
+- Column-delta cleaning / `original_col_id` — see [Forward compatibility](#forward-compatibility).
 - Multi-sort (sort by A then B). Single sort key only, matching today's behavior.
 - Persisting RowStore across widget re-renders. Each widget instance starts fresh.
diff --git a/packages/buckaroo-js-core/src/components/DFViewerParts/RowCacheController.test.ts b/packages/buckaroo-js-core/src/components/DFViewerParts/RowCacheController.test.ts
new file mode 100644
index 000000000..32c1530dd
--- /dev/null
+++ b/packages/buckaroo-js-core/src/components/DFViewerParts/RowCacheController.test.ts
@@ -0,0 +1,253 @@
+/**
+ * RowCacheController — the request/response lifecycle for the rowid-keyed
+ * cache. This is the JS-only orchestration layer (parallel to
+ * KeyAwareSmartRowCache; no cutover yet). These tests drive it against a
+ * mock reqFn that records outgoing requests; the test then feeds responses
+ * back in via addResponse to simulate the server.
+ *
+ * Lifecycle decisions encoded here:
+ *   - loading overlay during a gap = the success callback is NOT fired until
+ *     rows are actually present
+ *   - sort is two round-trips (sort -> rowidOrder -> rows), where the view
+ *     does not exist until the permutation lands
+ *   - an in-flight sort/filter is deduped; a second viewport request attaches
+ *   - after a sort lands, head/tail are eagerly prefetched (the "second
+ *     request" pattern)
+ *   - filter + sort compose client-side (filtered subset in sort order)
+ *   - rows under a non-identity view are fetched by rowid (only the missing
+ *     ones); the identity view fetches its window positionally via populate
+ */
+import { RowCacheController, RowCacheReq } from "./RowCacheController";
+import { DFDataRow } from "./DFWhole";
+
+const rowsFor = (rowids: number[]): DFDataRow[] =>
+    rowids.map((r) => ({ a: r, label: `row-${r}` } as unknown as DFDataRow));
+
+const aOf = (rows: DFDataRow[]): number[] => rows.map((r) => (r as unknown as { a: number }).a);
+
+function harness(opts: {
+    datasetLength: number;
+    headSize?: number;
+    tailSize?: number;
+    sortCapacity?: number;
+}) {
+    const sent: RowCacheReq[] = [];
+    const reqFn = (r: RowCacheReq) => sent.push(r);
+    const ctl = new RowCacheController(reqFn, {
+        sourceName: "s",
+        datasetLength: opts.datasetLength,
+        headSize: opts.headSize ?? 20,
+        tailSize: opts.tailSize ?? 20,
+        sortCapacity: opts.sortCapacity ?? 4,
+    });
+    const byKind = (k: string) => sent.filter((r) => r.kind === k);
+    return { sent, ctl, byKind };
+}
+
+const ident = (n: number) => Int32Array.from(Array.from({ length: n }, (_, i) => i));
+
+
+describe("RowCacheController — identity view", () => {
+    test("cold: parks the callback, fires a populate, resolves on the rows response", () => {
+        const { sent, ctl } = harness({ datasetLength: 100 });
+        const success = jest.fn();
+        const fail = jest.fn();
+        ctl.getRows({ sourceName: "s", start: 0, end: 5 }, success, fail);
+
+        // loading overlay: nothing resolved yet
+        expect(success).not.toHaveBeenCalled();
+        expect(sent).toEqual([
+            { kind: "populate", sourceName: "s", viewKey: "identity", start: 0, end: 5 },
+        ]);
+
+        ctl.addResponse({
+            kind: "rows",
+            sourceName: "s",
+            viewKey: "identity",
+            rowids: [0, 1, 2, 3, 4],
+            rows: rowsFor([0, 1, 2, 3, 4]),
+        });
+
+        expect(success).toHaveBeenCalledTimes(1);
+        expect(aOf(success.mock.calls[0][0])).toEqual([0, 1, 2, 3, 4]);
+        expect(success.mock.calls[0][1]).toBe(100);
+    });
+
+    test("warm: a cached window resolves synchronously with no new request", () => {
+        const { sent, ctl } = harness({ datasetLength: 100 });
+        ctl.getRows({ sourceName: "s", start: 0, end: 5 }, jest.fn(), jest.fn());
+        ctl.addResponse({
+            kind: "rows",
+            sourceName: "s",
+            viewKey: "identity",
+            rowids: [0, 1, 2, 3, 4],
+            rows: rowsFor([0, 1, 2, 3, 4]),
+        });
+        const before = sent.length;
+
+        const success = jest.fn();
+        ctl.getRows({ sourceName: "s", start: 0, end: 5 }, success, jest.fn());
+        expect(success).toHaveBeenCalledTimes(1);
+        expect(sent.length).toBe(before);
+    });
+});
+
+
+describe("RowCacheController — sort lifecycle", () => {
+    test("two round-trips: sort -> rowidOrder -> rowsByRowid -> rows, in sort order", () => {
+        const { sent, ctl } = harness({ datasetLength: 5, headSize: 0, tailSize: 0 });
+        const success = jest.fn();
+        ctl.getRows(
+            { sourceName: "s", sort: { sortKey: "age", sortDirection: "asc" }, start: 0, end: 3 },
+            success,
+            jest.fn(),
+        );
+
+        // first hop: a sort request only, no rows, no resolution
+        expect(success).not.toHaveBeenCalled();
+        expect(sent).toEqual([
+            { kind: "sort", sourceName: "s", sortKey: "age", sortDirection: "asc" },
+        ]);
+
+        ctl.addResponse({
+            kind: "sort",
+            sourceName: "s",
+            sortKey: "age",
+            sortDirection: "asc",
+            rowidOrder: Int32Array.from([4, 3, 2, 1, 0]),
+        });
+
+        // second hop: fetch the window's rowids by id; still no resolution
+        const rowsReq = sent.find((r) => r.kind === "rowsByRowid");
+        expect(rowsReq).toBeDefined();
+        expect((rowsReq as { rowids: number[] }).rowids).toEqual([4, 3, 2]);
+        expect(success).not.toHaveBeenCalled();
+
+        ctl.addResponse({
+            kind: "rows",
+            sourceName: "s",
+            viewKey: "sort:age:asc",
+            rowids: [4, 3, 2],
+            rows: rowsFor([4, 3, 2]),
+        });
+
+        expect(success).toHaveBeenCalledTimes(1);
+        expect(aOf(success.mock.calls[0][0])).toEqual([4, 3, 2]);
+        expect(success.mock.calls[0][1]).toBe(5);
+    });
+
+    test("dedupe: two viewport requests for the same unbuilt sort fire one sort request", () => {
+        const { ctl, byKind } = harness({ datasetLength: 10, headSize: 0, tailSize: 0 });
+        const s1 = jest.fn();
+        const s2 = jest.fn();
+        ctl.getRows({ sourceName: "s", sort: { sortKey: "age", sortDirection: "asc" }, start: 0, end: 3 }, s1, jest.fn());
+        ctl.getRows({ sourceName: "s", sort: { sortKey: "age", sortDirection: "asc" }, start: 3, end: 6 }, s2, jest.fn());
+
+        expect(byKind("sort").length).toBe(1);
+
+        ctl.addResponse({
+            kind: "sort", sourceName: "s", sortKey: "age", sortDirection: "asc",
+            rowidOrder: ident(10),
+        });
+        expect(byKind("rowsByRowid").length).toBe(2);
+
+        ctl.addResponse({ kind: "rows", sourceName: "s", viewKey: "sort:age:asc", rowids: [0, 1, 2], rows: rowsFor([0, 1, 2]) });
+        ctl.addResponse({ kind: "rows", sourceName: "s", viewKey: "sort:age:asc", rowids: [3, 4, 5], rows: rowsFor([3, 4, 5]) });
+        expect(s1).toHaveBeenCalledTimes(1);
+        expect(s2).toHaveBeenCalledTimes(1);
+    });
+
+    test("eager head/tail: after a sort lands, head and tail windows are prefetched", () => {
+        const { ctl, byKind } = harness({ datasetLength: 1000, headSize: 20, tailSize: 20 });
+        ctl.getRows(
+            { sourceName: "s", sort: { sortKey: "age", sortDirection: "asc" }, start: 500, end: 520 },
+            jest.fn(),
+            jest.fn(),
+        );
+        ctl.addResponse({
+            kind: "sort", sourceName: "s", sortKey: "age", sortDirection: "asc",
+            rowidOrder: ident(1000),
+        });
+
+        const fetches = byKind("rowsByRowid") as Array<{ rowids: number[] }>;
+        expect(fetches.some((r) => r.rowids[0] === 500)).toBe(true); // the visible window
+        expect(fetches.some((r) => r.rowids[0] === 0)).toBe(true); // head
+        expect(fetches.some((r) => r.rowids[r.rowids.length - 1] === 999)).toBe(true); // tail
+    });
+
+    test("scrolling under a built sort fetches only the missing rowids by id", () => {
+        const { sent, ctl } = harness({ datasetLength: 10, headSize: 0, tailSize: 0 });
+        ctl.getRows({ sourceName: "s", sort: { sortKey: "k", sortDirection: "asc" }, start: 0, end: 3 }, jest.fn(), jest.fn());
+        ctl.addResponse({ kind: "sort", sourceName: "s", sortKey: "k", sortDirection: "asc", rowidOrder: ident(10) });
+        ctl.addResponse({ kind: "rows", sourceName: "s", viewKey: "sort:k:asc", rowids: [0, 1, 2], rows: rowsFor([0, 1, 2]) });
+
+        const success = jest.fn();
+        ctl.getRows({ sourceName: "s", sort: { sortKey: "k", sortDirection: "asc" }, start: 2, end: 5 }, success, jest.fn());
+
+        const last = sent[sent.length - 1] as { kind: string; rowids: number[] };
+        expect(last.kind).toBe("rowsByRowid");
+        expect(last.rowids).toEqual([3, 4]); // 2 was already cached
+
+        ctl.addResponse({ kind: "rows", sourceName: "s", viewKey: "sort:k:asc", rowids: [3, 4], rows: rowsFor([3, 4]) });
+        expect(success).toHaveBeenCalledTimes(1);
+        expect(aOf(success.mock.calls[0][0])).toEqual([2, 3, 4]);
+    });
+});
+
+
+describe("RowCacheController — filter and composition", () => {
+    test("filter: builds the FilterView then fetches the subset rows", () => {
+        const { sent, ctl } = harness({ datasetLength: 100, headSize: 0, tailSize: 0 });
+        const success = jest.fn();
+        ctl.getRows({ sourceName: "s", filterKey: "age>50", start: 0, end: 3 }, success, jest.fn());
+
+        expect(sent).toEqual([{ kind: "filter", sourceName: "s", filterKey: "age>50" }]);
+
+        ctl.addResponse({ kind: "filter", sourceName: "s", filterKey: "age>50", rowidSubset: Int32Array.from([10, 20, 30, 40]) });
+        const rowsReq = sent.find((r) => r.kind === "rowsByRowid") as { rowids: number[] };
+        expect(rowsReq.rowids).toEqual([10, 20, 30]);
+
+        ctl.addResponse({ kind: "rows", sourceName: "s", viewKey: "filter:age>50", rowids: [10, 20, 30], rows: rowsFor([10, 20, 30]) });
+        expect(success).toHaveBeenCalledTimes(1);
+        expect(aOf(success.mock.calls[0][0])).toEqual([10, 20, 30]);
+        expect(success.mock.calls[0][1]).toBe(4); // filter length, not dataset length
+    });
+
+    test("filter + sort compose client-side: filtered subset in sort order, fetched by rowid", () => {
+        const { sent, ctl, byKind } = harness({ datasetLength: 10, headSize: 0, tailSize: 0 });
+        const success = jest.fn();
+        ctl.getRows(
+            { sourceName: "s", sort: { sortKey: "age", sortDirection: "desc" }, filterKey: "even", start: 0, end: 3 },
+            success,
+            jest.fn(),
+        );
+        expect(byKind("sort").length).toBe(1);
+        expect(byKind("filter").length).toBe(1);
+        expect(success).not.toHaveBeenCalled();
+
+        // descending order over 0..9
+        ctl.addResponse({ kind: "sort", sourceName: "s", sortKey: "age", sortDirection: "desc", rowidOrder: Int32Array.from([9, 8, 7, 6, 5, 4, 3, 2, 1, 0]) });
+        // evens, arbitrary order
+        ctl.addResponse({ kind: "filter", sourceName: "s", filterKey: "even", rowidSubset: Int32Array.from([0, 2, 4, 6, 8]) });
+
+        // composed = [8,6,4,2,0]; window [0,3) => [8,6,4]
+        const rowsReq = sent.find((r) => r.kind === "rowsByRowid") as { rowids: number[]; viewKey: string };
+        expect(rowsReq.rowids).toEqual([8, 6, 4]);
+
+        ctl.addResponse({ kind: "rows", sourceName: "s", viewKey: rowsReq.viewKey, rowids: [8, 6, 4], rows: rowsFor([8, 6, 4]) });
+        expect(success).toHaveBeenCalledTimes(1);
+        expect(aOf(success.mock.calls[0][0])).toEqual([8, 6, 4]);
+        expect(success.mock.calls[0][1]).toBe(5); // composed length
+    });
+});
+
+
+describe("RowCacheController — errors", () => {
+    test("an error fails the parked callbacks", () => {
+        const { ctl } = harness({ datasetLength: 100, headSize: 0, tailSize: 0 });
+        const fail = jest.fn();
+        ctl.getRows({ sourceName: "s", start: 0, end: 5 }, jest.fn(), fail);
+        ctl.addError();
+        expect(fail).toHaveBeenCalledTimes(1);
+    });
+});
diff --git a/packages/buckaroo-js-core/src/components/DFViewerParts/RowCacheController.ts b/packages/buckaroo-js-core/src/components/DFViewerParts/RowCacheController.ts
new file mode 100644
index 000000000..776930934
--- /dev/null
+++ b/packages/buckaroo-js-core/src/components/DFViewerParts/RowCacheController.ts
@@ -0,0 +1,255 @@
+import { DFDataRow } from "./DFWhole";
+import { RowCache, RowCacheConfig } from "./RowCache";
+import {
+    View,
+    SortView,
+    FilterView,
+    SortDirection,
+    sortViewKey,
+    filterViewKey,
+    IDENTITY_VIEW_KEY,
+} from "./Views";
+
+/**
+ * RowCacheController — the request/response lifecycle for the rowid-keyed
+ * cache (see docs/smart-row-cache-redesign.md, phase 7b).
+ *
+ * This is the orchestration layer that `RowCache` (pure state) lacks: it
+ * fires requests, parks the AG-Grid callback, replays it when the data
+ * lands, dedupes in-flight work, and eagerly prefetches head/tail. It is
+ * built net-new and runs PARALLEL to `KeyAwareSmartRowCache` — nothing is
+ * cut over until the rowid system is ready to merge. It is JS-only and
+ * fully testable against a mock `reqFn` (see RowCacheController.test.ts).
+ *
+ * One controller owns one generation (one `original_row_id` namespace).
+ * The `Map<generationKey, RowCacheController>` wrapper is a follow-on
+ * (see "Decisions resolved → the cache key is the generation").
+ *
+ * Lifecycle (reactive `drive()` loop):
+ *   resolve the active view — if its permutation/subset isn't here yet, fire
+ *   a `sort`/`filter` request and park (AG-Grid shows its loading overlay
+ *   because we don't call the callback). Once built, `missingAt` the window;
+ *   if present, resolve + `gc`; if not, fetch the missing rowids (by rowid,
+ *   or positionally via `populate` for the identity view) and park. Every
+ *   response re-drives all parked requests, so partial overlap and
+ *   filter+sort composition fall out naturally.
+ */
+
+export type RowCacheReq =
+    | { kind: "sort"; sourceName: string; sortKey: string; sortDirection: SortDirection }
+    | { kind: "filter"; sourceName: string; filterKey: string }
+    | { kind: "rowsByRowid"; sourceName: string; viewKey: string; rowids: number[] }
+    | { kind: "populate"; sourceName: string; viewKey: string; start: number; end: number };
+
+export type RowCacheResp =
+    | { kind: "sort"; sourceName: string; sortKey: string; sortDirection: SortDirection; rowidOrder: Int32Array }
+    | { kind: "filter"; sourceName: string; filterKey: string; rowidSubset: Int32Array }
+    | { kind: "rows"; sourceName: string; viewKey: string; rowids: number[]; rows: DFDataRow[] };
+
+export interface GridReq {
+    sourceName: string;
+    sort?: { sortKey: string; sortDirection: SortDirection } | null;
+    filterKey?: string | null;
+    start: number;
+    end: number;
+}
+
+export type SuccessCB = (rows: DFDataRow[], length: number) => void;
+export type FailCB = () => void;
+export type ReqFN = (req: RowCacheReq) => void;
+
+export type RowCacheControllerConfig = RowCacheConfig & { sourceName: string };
+
+interface Pending {
+    req: GridReq;
+    success: SuccessCB;
+    fail: FailCB;
+}
+
+
+export class RowCacheController {
+    private readonly cache: RowCache;
+    private readonly reqFn: ReqFN;
+    private readonly sourceName: string;
+    private readonly headSize: number;
+    private readonly tailSize: number;
+
+    private pending: Pending[] = [];
+    // viewKeys of sort/filter permutations currently being fetched
+    private readonly inFlightViews = new Set<string>();
+    // signatures of row fetches currently outstanding
+    private readonly inFlightFetches = new Set<string>();
+
+    constructor(reqFn: ReqFN, cfg: RowCacheControllerConfig) {
+        this.reqFn = reqFn;
+        this.sourceName = cfg.sourceName;
+        this.cache = new RowCache(cfg);
+        const resolved = this.cache.config();
+        this.headSize = resolved.headSize;
+        this.tailSize = resolved.tailSize;
+    }
+
+    /** AG-Grid datasource entry point. Resolves async; `success` fires only
+     * once the rows are present (loading overlay shows until then). */
+    public getRows(req: GridReq, success: SuccessCB, fail: FailCB): void {
+        this.pending.push({ req, success, fail });
+        this.drive();
+    }
+
+    /** Feed a server response back in. */
+    public addResponse(resp: RowCacheResp): void {
+        if (resp.kind === "sort") {
+            const view = this.cache.applySort({
+                sortKey: resp.sortKey,
+                sortDirection: resp.sortDirection,
+                rowidOrder: resp.rowidOrder,
+            });
+            this.inFlightViews.delete(sortViewKey(resp.sortKey, resp.sortDirection));
+            this.drive();
+            this.eagerHeadTail(view);
+        } else if (resp.kind === "filter") {
+            this.cache.applyFilter({ filterKey: resp.filterKey, rowidSubset: resp.rowidSubset });
+            this.inFlightViews.delete(filterViewKey(resp.filterKey));
+            this.drive();
+        } else {
+            // Assumes the response carries every requested rowid (complete).
+            this.cache.populate({ rowids: resp.rowids, rows: resp.rows });
+            this.inFlightFetches.delete(this.sigForRows(resp.viewKey, resp.rowids));
+            this.drive();
+        }
+    }
+
+    /** Server-side failure: fail every parked callback and reset in-flight. */
+    public addError(): void {
+        const parked = this.pending;
+        this.pending = [];
+        this.inFlightViews.clear();
+        this.inFlightFetches.clear();
+        for (const p of parked) p.fail();
+    }
+
+    public pendingCount(): number {
+        return this.pending.length;
+    }
+
+    public rowCache(): RowCache {
+        return this.cache;
+    }
+
+    // ---- internals -------------------------------------------------------
+
+    private drive(): void {
+        const still: Pending[] = [];
+        for (const p of this.pending) {
+            if (!this.tryResolve(p)) still.push(p);
+        }
+        this.pending = still;
+    }
+
+    /** Returns true if `p` was satisfied (success fired), false if parked. */
+    private tryResolve(p: Pending): boolean {
+        const view = this.resolveViewOrRequest(p.req);
+        if (view === undefined) return false;
+        const { start, end } = p.req;
+        const missing = this.cache.missingAt(view, start, end);
+        if (missing.length === 0) {
+            const rows = this.cache.rowsAt(view, start, end) as DFDataRow[];
+            p.success(rows, view.length());
+            this.cache.gc({ view, start, end });
+            return true;
+        }
+        this.fetchMissing(view, start, end, missing);
+        return false;
+    }
+
+    /** Resolve the view for a request; if a sort/filter permutation is not
+     * built yet, fire the request(s) (deduped) and return undefined. */
+    private resolveViewOrRequest(req: GridReq): View | undefined {
+        const hasSort = req.sort !== undefined && req.sort !== null;
+        const hasFilter = req.filterKey !== undefined && req.filterKey !== null;
+        if (!hasSort && !hasFilter) return this.cache.defaultView();
+
+        let sortView: SortView | undefined;
+        let filterView: FilterView | undefined;
+        if (hasSort) {
+            sortView = this.cache.getSortView(req.sort!.sortKey, req.sort!.sortDirection);
+            if (sortView === undefined) this.requestSort(req.sort!.sortKey, req.sort!.sortDirection);
+        }
+        if (hasFilter) {
+            filterView = this.cache.getFilterView(req.filterKey!);
+            if (filterView === undefined) this.requestFilter(req.filterKey!);
+        }
+        if (hasSort && hasFilter) {
+            if (sortView !== undefined && filterView !== undefined) {
+                return this.composeView(sortView, filterView);
+            }
+            return undefined;
+        }
+        return hasSort ? sortView : filterView;
+    }
+
+    /** Filtered subset in sort order: the filtered rowids, ordered by the
+     * sort permutation. Free because both are keyed off original_row_id. */
+    private composeView(s: SortView, f: FilterView): FilterView {
+        const subset = new Set<number>();
+        for (let i = 0; i < f.rowidSubset.length; i++) subset.add(f.rowidSubset[i]);
+        const out: number[] = [];
+        for (let i = 0; i < s.rowidOrder.length; i++) {
+            const r = s.rowidOrder[i];
+            if (subset.has(r)) out.push(r);
+        }
+        return new FilterView(`${s.viewKey()}&${f.viewKey()}`, Int32Array.from(out));
+    }
+
+    private requestSort(sortKey: string, sortDirection: SortDirection): void {
+        const vk = sortViewKey(sortKey, sortDirection);
+        if (this.inFlightViews.has(vk)) return;
+        this.inFlightViews.add(vk);
+        this.reqFn({ kind: "sort", sourceName: this.sourceName, sortKey, sortDirection });
+    }
+
+    private requestFilter(filterKey: string): void {
+        const vk = filterViewKey(filterKey);
+        if (this.inFlightViews.has(vk)) return;
+        this.inFlightViews.add(vk);
+        this.reqFn({ kind: "filter", sourceName: this.sourceName, filterKey });
+    }
+
+    private fetchMissing(view: View, start: number, end: number, missing: number[]): void {
+        const vk = view.viewKey();
+        if (vk === IDENTITY_VIEW_KEY) {
+            // identity is positional: fetch the whole window via populate
+            const windowRowids = view.rowidsInRange(start, end);
+            const sig = this.sigForRows(vk, windowRowids);
+            if (this.inFlightFetches.has(sig)) return;
+            this.inFlightFetches.add(sig);
+            this.reqFn({ kind: "populate", sourceName: this.sourceName, viewKey: vk, start, end });
+        } else {
+            // non-identity: fetch only the missing rowids by id
+            const sig = this.sigForRows(vk, missing);
+            if (this.inFlightFetches.has(sig)) return;
+            this.inFlightFetches.add(sig);
+            this.reqFn({ kind: "rowsByRowid", sourceName: this.sourceName, viewKey: vk, rowids: missing });
+        }
+    }
+
+    /** Eagerly prefetch head/tail of a freshly-built view so the GC pin has
+     * something to hold (the "second request" pattern). Fire-and-forget. */
+    private eagerHeadTail(view: View): void {
+        const len = view.length();
+        const windows: Array<[number, number]> = [];
+        if (this.headSize > 0) windows.push([0, Math.min(this.headSize, len)]);
+        if (this.tailSize > 0) windows.push([Math.max(0, len - this.tailSize), len]);
+        for (const [s, e] of windows) {
+            if (e <= s) continue;
+            const missing = this.cache.missingAt(view, s, e);
+            if (missing.length === 0) continue;
+            this.fetchMissing(view, s, e, missing);
+        }
+    }
+
+    private sigForRows(viewKey: string, rowids: number[] | Int32Array): string {
+        const arr = Array.from(rowids).sort((a, b) => a - b);
+        return `${viewKey}:${arr.join(",")}`;
+    }
+}