diff --git a/src/window.zig b/src/window.zig
index 253077a..3218a5c 100644
--- a/src/window.zig
+++ b/src/window.zig
@@ -152,3 +152,157 @@ fn isFrameValue(kind: WindowKind) bool {
     return kind == .first_value or kind == .last_value or kind == .nth_value;
 }
 
+/// The row field name a `compute()` result column is exposed under (e.g.
+/// for CLI/protocol output) - bare `WindowKind` name for the ranking family
+/// and for `count`'s `COUNT(*)` form (`field.len == 0`), otherwise
+/// `"kind_field"`. Format-identical to `aggregate.zig`'s `specLabel`.
+pub fn specLabel(allocator: Allocator, spec: WindowSpec) Allocator.Error![]u8 {
+    if (spec.field.len == 0) return allocator.dupe(u8, @tagName(spec.kind));
+    return std.fmt.allocPrint(allocator, "{s}_{s}", .{ @tagName(spec.kind), spec.field });
+}
+
+/// Streaming per-row accumulator for the running-aggregate family
+/// (`sum`/`avg`/`count`/`min`/`max`). Mirrors `aggregate.zig`'s `AccState`
+/// (same int-exact/float-fallback dual for `sum`/`avg`, same
+/// compare-and-replace for `min`/`max`), minus `count_distinct` (out of
+/// scope here - a running "distinct count so far" is a real SQL feature but
+/// a separate piece of work). The key behavioral difference from
+/// `AccState.take()`: `snapshot()` does NOT transfer ownership of
+/// `minv`/`maxv` out of the accumulator, because a running window emits a
+/// result after EVERY row, not once at the very end - the accumulator must
+/// keep its own copy alive for the next row's snapshot too.
+const RunningAcc = struct {
+    kind: WindowKind,
+    field: []const u8,
+    n: u64 = 0,
+    sumInt: i64 = 0,
+    sumFloat: f64 = 0,
+    allInt: bool = true,
+    minv: ?Value = null,
+    maxv: ?Value = null,
+    /// `.count_distinct` only: encoded-`Value` bytes seen so far, one entry
+    /// per distinct value - same encode-as-hashmap-key trick
+    /// `aggregate.zig`'s `AccState.distinct` uses. Unlike that GROUP BY
+    /// counterpart, `snapshot()` below never drains this (a running window
+    /// needs the set to keep growing across the WHOLE partition walk, not
+    /// just until one final result) - only `deinit` ever frees it.
+    distinct: std.StringHashMapUnmanaged(void) = .{},
+
+    fn observe(self: *RunningAcc, allocator: Allocator, row: *const Row) WindowError!void {
+        switch (self.kind) {
+            .count => {
+                if (self.field.len == 0 or row.get(self.field) != null) self.n += 1;
+            },
+            .count_distinct => {
+                const v = row.get(self.field) orelse return;
+                const key = try rowmod.encodeValue(allocator, v);
+                const res = try self.distinct.getOrPut(allocator, key);
+                if (res.found_existing) allocator.free(key);
+            },
+            .sum, .avg => {
+                const v = row.get(self.field) orelse return;
+                switch (v) {
+                    .int => |i| {
+                        if (self.allInt) {
+                            self.sumInt = std.math.add(i64, self.sumInt, i) catch return error.Overflow;
+                        } else {
+                            self.sumFloat += @floatFromInt(i);
+                        }
+                    },
+                    .float => |f| {
+                        if (self.allInt) {
+                            // First float observed after an int-only run:
+                            // migrate the exact integer sum so far into the
+                            // float accumulator once, then continue in float.
+                            self.sumFloat = @floatFromInt(self.sumInt);
+                            self.allInt = false;
+                        }
+                        self.sumFloat += f;
+                    },
+                    else => return error.NonNumericField,
+                }
+                self.n += 1;
+            },
+            .min => {
+                const v = row.get(self.field) orelse return;
+                if (self.minv == null or rowmod.compare(v, self.minv.?) == .lt) {
+                    if (self.minv) |old| old.free(allocator);
+                    self.minv = try v.dupe(allocator);
+                }
+            },
+            .max => {
+                const v = row.get(self.field) orelse return;
+                if (self.maxv == null or rowmod.compare(v, self.maxv.?) == .gt) {
+                    if (self.maxv) |old| old.free(allocator);
+                    self.maxv = try v.dupe(allocator);
+                }
+            },
+            .row_number, .rank, .dense_rank, .lag, .lead, .first_value, .last_value, .nth_value => unreachable,
+        }
+    }
+
+    /// Returns the accumulator's CURRENT state as a freshly owned `Value`
+    /// (a dupe for `min`/`max`, since the accumulator keeps its own copy
+    /// alive for subsequent rows - see the type doc comment above).
+    /// `.count_distinct` is likewise non-draining - `self.distinct.count()`
+    /// is read, never emptied, so the NEXT row's snapshot still sees every
+    /// value observed so far (only `deinit` ever frees the set).
+    fn snapshot(self: *const RunningAcc, allocator: Allocator) Allocator.Error!Value {
+        return switch (self.kind) {
+            .count => .{ .int = @intCast(self.n) },
+            .count_distinct => .{ .int = @intCast(self.distinct.count()) },
+            .sum => if (self.allInt) .{ .int = self.sumInt } else .{ .float = self.sumFloat },
+            .avg => blk: {
+                if (self.n == 0) break :blk .{ .none = {} };
+                const total: f64 = if (self.allInt) @floatFromInt(self.sumInt) else self.sumFloat;
+                break :blk .{ .float = total / @as(f64, @floatFromInt(self.n)) };
+            },
+            .min => if (self.minv) |v| try v.dupe(allocator) else .{ .none = {} },
+            .max => if (self.maxv) |v| try v.dupe(allocator) else .{ .none = {} },
+            .row_number, .rank, .dense_rank, .lag, .lead, .first_value, .last_value, .nth_value => unreachable,
+        };
+    }
+
+    fn deinit(self: *RunningAcc, allocator: Allocator) void {
+        if (self.minv) |v| v.free(allocator);
+        if (self.maxv) |v| v.free(allocator);
+        var it = self.distinct.keyIterator();
+        while (it.next()) |k| allocator.free(k.*);
+        self.distinct.deinit(allocator);
+    }
+};
+
+/// Groups row indices `0..rows.len` by `partitionBy` (empty = a single
+/// partition covering every row), preserving first-seen partition order -
+/// the window-function analogue of `aggregate.zig`'s `GroupBy` grouping,
+/// except every member index is kept instead of collapsed into one result.
+/// Multi-column partition keys are built with `row.appendOrdered`
+/// (concatenation-safe and unambiguous by construction), the same
+/// primitive `index.zig` uses for composite index keys - a row missing one
+/// of the `partitionBy` fields contributes `.none` for that column rather
+/// than being excluded (unlike `aggregate.zig`'s GROUP BY, which drops rows
+/// missing the group field entirely: a window function must still annotate
+/// every input row with SOME result, so there is no row to drop it from).
+/// Caller owns the returned slice of slices AND each inner slice
+/// (`allocator.free` each inner slice, then the outer one).
+fn buildPartitions(allocator: Allocator, rows: []const Row, partitionBy: []const []const u8) Allocator.Error![][]usize {
+    var order: std.ArrayListUnmanaged([]const u8) = .{};
+    defer {
+        for (order.items) |k| allocator.free(k);
+        order.deinit(allocator);
+    }
+    var map: std.StringHashMapUnmanaged(std.ArrayListUnmanaged(usize)) = .{};
+    defer {
+        var it = map.iterator();
+        while (it.next()) |e| e.value_ptr.deinit(allocator);
+        map.deinit(allocator);
+    }
+
+    for (rows, 0..) |*r, i| {
+        var keyBuf = std.ArrayList(u8).init(allocator);
+        defer keyBuf.deinit();
+        for (partitionBy) |f| {
+            const v = r.get(f) orelse Value{ .none = {} };
+            try rowmod.appendOrdered(&keyBuf, v);
+        }
+