diff --git a/src/protocol.zig b/src/protocol.zig
index b05ec99..1eabdbd 100644
--- a/src/protocol.zig
+++ b/src/protocol.zig
@@ -649,3 +649,155 @@ pub const OwnedCollReply = union(enum) {
     pairList: []OwnedCollPair,
     scorePairList: []OwnedCollScorePair,
 
+    pub fn deinit(self: *OwnedCollReply, allocator: Allocator) void {
+        switch (self.*) {
+            .bytes => |b| allocator.free(b),
+            .stringList => |items| {
+                for (items) |it| allocator.free(it);
+                allocator.free(items);
+            },
+            .pairList => |items| {
+                for (items) |p| {
+                    allocator.free(p.field);
+                    allocator.free(p.value);
+                }
+                allocator.free(items);
+            },
+            .scorePairList => |items| {
+                for (items) |p| allocator.free(p.member);
+                allocator.free(items);
+            },
+            .nil, .ok, .int, .real => {},
+        }
+        self.* = undefined;
+    }
+};
+
+pub fn decodeCollReply(allocator: Allocator, payload: []const u8) !OwnedCollReply {
+    var c = Cursor{ .data = payload };
+    const tag = std.meta.intToEnum(CollReplyTag, try c.takeByte()) catch return error.Malformed;
+    return switch (tag) {
+        .nil => .nil,
+        .ok => .ok,
+        .int => .{ .int = @bitCast(try c.takeU64()) },
+        .real => .{ .real = @bitCast(try c.takeU64()) },
+        .bytes => .{ .bytes = try allocator.dupe(u8, c.data[c.pos..]) },
+        .stringList => blk: {
+            const count = try c.takeU32();
+            var list = try std.ArrayList([]u8).initCapacity(allocator, count);
+            errdefer {
+                for (list.items) |it| allocator.free(it);
+                list.deinit();
+            }
+            var i: u32 = 0;
+            while (i < count) : (i += 1) {
+                const raw = try c.takeLenPrefixed();
+                list.appendAssumeCapacity(try allocator.dupe(u8, raw));
+            }
+            break :blk .{ .stringList = try list.toOwnedSlice() };
+        },
+        .pairList => blk: {
+            const count = try c.takeU32();
+            var list = try std.ArrayList(OwnedCollPair).initCapacity(allocator, count);
+            errdefer {
+                for (list.items) |p| {
+                    allocator.free(p.field);
+                    allocator.free(p.value);
+                }
+                list.deinit();
+            }
+            var i: u32 = 0;
+            while (i < count) : (i += 1) {
+                const field = try c.takeLenPrefixed();
+                const value = try c.takeLenPrefixed();
+                list.appendAssumeCapacity(.{ .field = try allocator.dupe(u8, field), .value = try allocator.dupe(u8, value) });
+            }
+            break :blk .{ .pairList = try list.toOwnedSlice() };
+        },
+        .scorePairList => blk: {
+            const count = try c.takeU32();
+            var list = try std.ArrayList(OwnedCollScorePair).initCapacity(allocator, count);
+            errdefer {
+                for (list.items) |p| allocator.free(p.member);
+                list.deinit();
+            }
+            var i: u32 = 0;
+            while (i < count) : (i += 1) {
+                const member = try c.takeLenPrefixed();
+                const score: f64 = @bitCast(try c.takeU64());
+                list.appendAssumeCapacity(.{ .member = try allocator.dupe(u8, member), .score = score });
+            }
+            break :blk .{ .scorePairList = try list.toOwnedSlice() };
+        },
+    };
+}
+
+// --- txn_op ---
+
+/// Remote counterpart of the CLI's local `txn <script-file>` (main.zig's
+/// `runTxn`, backed by `txn.zig`/`txn_table.zig`): a batch of put/delete row
+/// operations, possibly spanning several tables, applied as one atomic
+/// `Transaction` commit - the same "every op succeeds together or none of
+/// them do" guarantee the local CLI gives, now reachable from a process that
+/// only has a `Client`, not a `*Db`. Unlike `coll_op`/`ttl_op` (one action
+/// against one key/collection per request) this opcode carries a LIST of
+/// entries, since a transaction script is inherently a batch.
+pub const TxnVerb = enum(u8) {
+    put = 0,
+    del = 1,
+
+    pub fn fromByte(b: u8) !TxnVerb {
+        return std.meta.intToEnum(TxnVerb, b) catch error.Malformed;
+    }
+};
+
+/// One staged operation as the caller supplies it. `fields` is ignored for
+/// `.del` (callers pass `&.{}`) - same "unused argument still there, just
+/// empty" convention as `encodeUserAdmin`'s zeroed `role`/`rounds` on
+/// actions that do not need them.
+pub const TxnOpEntry = struct {
+    verb: TxnVerb,
+    table: []const u8,
+    pk: rowmod.Value,
+    fields: []const rowmod.Field = &.{},
+};
+
+pub fn encodeTxnOp(allocator: Allocator, entries: []const TxnOpEntry) ![]u8 {
+    var buf = std.ArrayList(u8).init(allocator);
+    errdefer buf.deinit();
+    const w = encode.ByteWriter.init(&buf);
+    try w.putU32(@intCast(entries.len));
+    for (entries) |e| {
+        try w.byte(@intFromEnum(e.verb));
+        try putLenPrefixed(w, e.table);
+
+        const pkBytes = try rowmod.encodeValue(allocator, e.pk);
+        defer allocator.free(pkBytes);
+        try putLenPrefixed(w, pkBytes);
+
+        // Reuses `Row.encode` for the field list exactly the way
+        // `encodeRows` does for a whole result set, instead of inventing a
+        // second name+value framing just for this opcode.
+        var fieldsRow = rowmod.Row.init(allocator);
+        defer fieldsRow.deinit();
+        for (e.fields) |f| try fieldsRow.set(f.name, f.value);
+        const fieldsBytes = try fieldsRow.encode(allocator);
+        defer allocator.free(fieldsBytes);
+        try putLenPrefixed(w, fieldsBytes);
+    }
+    return buf.toOwnedSlice();
+}
+
+/// Decoded entry with every slice borrowed from the frame `payload` that
+/// carried it - same convention as `decodeCollOp`/`decodeTableSelect`, valid
+/// only as long as that payload is alive. `pkBytes`/`fieldsBytes` still need
+/// `rowmod.decodeValue`/`rowmod.Row.decode` to become a real `Value`/field
+/// list; kept undecoded here so a malformed table name or verb byte fails
+/// fast without touching row-decoding at all.
+pub const DecodedTxnOpEntry = struct {
+    verb: TxnVerb,
+    table: []const u8,
+    pkBytes: []const u8,
+    fieldsBytes: []const u8,
+};
+