diff --git a/src/protocol.zig b/src/protocol.zig
index 52fecc4..e4c55f8 100644
--- a/src/protocol.zig
+++ b/src/protocol.zig
@@ -1261,3 +1261,161 @@ pub fn freeJoinPairs(allocator: Allocator, pairs: []JoinPair) void {
 
 // --- table aggregate ---
 
+/// Wire request for a remote `table aggregate`: a table name, an optional
+/// GROUP BY field (empty means no grouping - the network counterpart of the
+/// CLI's `*` sentinel for `table aggregate <table> *`, chosen client-side so
+/// the wire itself needs no separate mode byte), the list of
+/// `aggregate.zig` specs to compute, and an `expr.zig` predicate text
+/// filtering rows BEFORE aggregation runs - the same empty-means-every-row
+/// convention `table_select`'s `exprText` already uses.
+/// `offset`/`limit` window the result GROUPS (not the input rows) when
+/// `groupField` is non-empty, the same `join.applyPage`/`Table.selectPage`
+/// convention as `table_join`'s wire format - 0/0 means unlimited. Ignored
+/// server-side when `groupField` is empty (`runGlobal` always produces
+/// exactly one result row, nothing to page).
+/// `orderBy` is a `query.OrderBy` sort key list (possibly empty) applied
+/// server-side to the result GROUPS via `aggregate.sortGroupsBy`, after
+/// HAVING narrows them and before `offset`/`limit` window them - same
+/// convention and wire shape (count byte + descending byte + length-prefixed
+/// field name, per column) as `encodeTableSelect`'s `orderBy`. Each
+/// `orderBy[i].field` must name either `groupField` or one of `specs`'
+/// `specLabel`s (validated server-side by `sortGroupsBy`, not here).
+/// Meaningless (and ignored server-side) for the ungrouped `groupField.len
+/// == 0` path, same as `offset`/`limit` already are - `runGlobal` always
+/// produces exactly one result row, nothing to order.
+///
+/// `havingText` is a second, independent `expr.zig` predicate text
+/// (empty means no HAVING) filtering the result GROUPS AFTER aggregation,
+/// evaluated server-side before `offset`/`limit` window them - the wire
+/// counterpart of the CLI's `having <expr>` suffix. Encoded length-prefixed
+/// (not "rest of payload" like `exprText`) precisely because `exprText`
+/// still needs to occupy that final slot.
+pub fn encodeTableAggregate(allocator: Allocator, table: []const u8, groupField: []const u8, specs: []const aggregatemod.AggSpec, offset: u32, limit: u32, orderBy: []const querymod.OrderBy, havingText: []const u8, exprText: []const u8) ![]u8 {
+    var buf = std.ArrayList(u8).init(allocator);
+    errdefer buf.deinit();
+    const w = encode.ByteWriter.init(&buf);
+    try putLenPrefixed(w, table);
+    try putLenPrefixed(w, groupField);
+    try w.putU32(@intCast(specs.len));
+    for (specs) |spec| {
+        try w.byte(@intFromEnum(spec.kind));
+        try putLenPrefixed(w, spec.field);
+    }
+    try w.putU32(offset);
+    try w.putU32(limit);
+    try w.byte(@intCast(orderBy.len));
+    for (orderBy) |ob| {
+        try w.byte(if (ob.descending) 1 else 0);
+        try putLenPrefixed(w, ob.field);
+    }
+    try putLenPrefixed(w, havingText);
+    try w.slice(exprText);
+    return buf.toOwnedSlice();
+}
+
+pub const TableAggregateArgs = struct {
+    table: []const u8,
+    groupField: []const u8,
+    /// Borrows `field` slices from the decoded payload but owns the slice
+    /// itself - caller frees with `allocator.free(args.specs)`, same shape
+    /// as `decodeBatch`'s returned list.
+    specs: []aggregatemod.AggSpec,
+    offset: u32 = 0,
+    limit: u32 = 0,
+    /// Empty means no ORDER BY - see `encodeTableAggregate`'s doc. Borrows
+    /// `field` slices from the decoded payload but owns the slice itself,
+    /// same convention as `specs`/`TableSelectArgs.orderBy`.
+    orderBy: []const querymod.OrderBy = &.{},
+    /// Empty means no HAVING clause - see `encodeTableAggregate`'s doc.
+    havingText: []const u8,
+    exprText: []const u8,
+};
+
+pub fn decodeTableAggregate(allocator: Allocator, payload: []const u8) !TableAggregateArgs {
+    var c = Cursor{ .data = payload };
+    const table = try c.takeLenPrefixed();
+    const groupField = try c.takeLenPrefixed();
+    const count = try c.takeU32();
+    var specs = try std.ArrayList(aggregatemod.AggSpec).initCapacity(allocator, count);
+    errdefer specs.deinit();
+    var i: u32 = 0;
+    while (i < count) : (i += 1) {
+        const kindByte = try c.takeByte();
+        const kind = std.meta.intToEnum(aggregatemod.AggKind, kindByte) catch return error.Malformed;
+        const field = try c.takeLenPrefixed();
+        specs.appendAssumeCapacity(.{ .kind = kind, .field = field });
+    }
+    const offset = try c.takeU32();
+    const limit = try c.takeU32();
+    const orderCount = try c.takeByte();
+    var orderBy = try std.ArrayList(querymod.OrderBy).initCapacity(allocator, orderCount);
+    errdefer orderBy.deinit();
+    var j: u8 = 0;
+    while (j < orderCount) : (j += 1) {
+        const descending = try c.takeByte();
+        const field = try c.takeLenPrefixed();
+        orderBy.appendAssumeCapacity(.{ .field = field, .descending = descending != 0 });
+    }
+    const havingText = try c.takeLenPrefixed();
+    const exprText = c.data[c.pos..];
+    return .{ .table = table, .groupField = groupField, .specs = try specs.toOwnedSlice(), .offset = offset, .limit = limit, .orderBy = try orderBy.toOwnedSlice(), .havingText = havingText, .exprText = exprText };
+}
+
+fn putEncodedValue(allocator: Allocator, w: encode.ByteWriter, value: rowmod.Value) !void {
+    const encoded = try rowmod.encodeValue(allocator, value);
+    defer allocator.free(encoded);
+    try putLenPrefixed(w, encoded);
+}
+
+fn takeDecodedValue(allocator: Allocator, c: *Cursor) !rowmod.Value {
+    const raw = try c.takeLenPrefixed();
+    return rowmod.decodeValue(allocator, raw) catch return error.Malformed;
+}
+
+/// Encodes a `runGlobal` (no `GROUP BY`) result: exactly `specs.len` values,
+/// same order as the request's specs.
+pub fn encodeAggGlobalResult(allocator: Allocator, values: []const rowmod.Value) ![]u8 {
+    var buf = std.ArrayList(u8).init(allocator);
+    errdefer buf.deinit();
+    const w = encode.ByteWriter.init(&buf);
+    try w.putU32(@intCast(values.len));
+    for (values) |v| try putEncodedValue(allocator, w, v);
+    return buf.toOwnedSlice();
+}
+
+/// Decodes a `runGlobal` result into independently owned `Value`s - callers
+/// must free with `freeAggValues`.
+pub fn decodeAggGlobalResult(allocator: Allocator, payload: []const u8) ![]rowmod.Value {
+    var c = Cursor{ .data = payload };
+    const count = try c.takeU32();
+    var list = try std.ArrayList(rowmod.Value).initCapacity(allocator, count);
+    errdefer {
+        for (list.items) |v| v.free(allocator);
+        list.deinit();
+    }
+    var i: u32 = 0;
+    while (i < count) : (i += 1) list.appendAssumeCapacity(try takeDecodedValue(allocator, &c));
+    return list.toOwnedSlice();
+}
+
+pub fn freeAggValues(allocator: Allocator, values: []rowmod.Value) void {
+    for (values) |v| v.free(allocator);
+    allocator.free(values);
+}
+
+/// Encodes a grouped `run` result: one entry per `GroupRow` (the group-key
+/// value, then exactly `specs.len` result values), in the same first-seen
+/// order `GroupBy.finish()` produced them.
+pub fn encodeAggGroupRows(allocator: Allocator, groups: []const aggregatemod.GroupRow) ![]u8 {
+    var buf = std.ArrayList(u8).init(allocator);
+    errdefer buf.deinit();
+    const w = encode.ByteWriter.init(&buf);
+    try w.putU32(@intCast(groups.len));
+    for (groups) |g| {
+        try putEncodedValue(allocator, w, g.group);
+        try w.putU32(@intCast(g.values.len));
+        for (g.values) |v| try putEncodedValue(allocator, w, v);
+    }
+    return buf.toOwnedSlice();
+}
+