13 files changed, 1133 insertions(+), 2 deletions(-)

M cmd/protoc-gen-tarantool/internal/gen/gen.go
M cmd/protoc-gen-tarantool/internal/gen/inline.go
M examples/expected/full/conformance/conformance_pb.lua
M examples/expected/full/hello/hello_pb.lua
M examples/expected/full/protobuf_test_messages/proto3/test_messages_proto3_pb.lua
M examples/expected/runtime/conformance/conformance_pb.lua
M examples/expected/runtime/hello/hello_pb.lua
M examples/expected/runtime/protobuf_test_messages/proto3/test_messages_proto3_pb.lua
M runtime/pb/codec.lua
M runtime/pb/init.lua
A runtime/pb/lazy.lua
M test/interop_test.lua
A test/lazy_test.lua

@@ 378,6 378,7 @@ func emitMessageWrappers(w *writer, file *protogen.File, m *protogen.Message) {
 	w.line("function M.%s_new(t) return t or {} end", name)
 	w.line("function M.%s_encode(t) return pb.encode(M.%s_descriptor, t) end", name, name)
 	w.line("function M.%s_decode(b) return pb.decode(M.%s_descriptor, b) end", name, name)
+	w.line("function M.%s_decode_lazy(b) return pb.decode_lazy(M.%s_descriptor, b) end", name, name)
 	emitOptionalAccessors(w, name, m)
 	w.line("")
 }

@@ 20,6 20,7 @@ func emitInlineMessage(w *writer, file *protogen.File, m *protogen.Message, impo
 
 	emitInlineEncode(w, name, m, file, selfPath, imports, prefix)
 	emitInlineDecode(w, name, m, file, selfPath, imports, prefix)
+	w.line("function M.%s_decode_lazy(b) return pb.decode_lazy(M.%s_descriptor, b) end", name, name)
 	emitOptionalAccessors(w, name, m)
 	w.line("")
 }

@@ 154,6 154,7 @@ function M.TestStatus_decode(buf)
     return result
 end
 
+function M.TestStatus_decode_lazy(b) return pb.decode_lazy(M.TestStatus_descriptor, b) end
 
 function M.FailureSet_new(t) return t or {} end
 


@@ 204,6 205,7 @@ function M.FailureSet_decode(buf)
     return result
 end
 
+function M.FailureSet_decode_lazy(b) return pb.decode_lazy(M.FailureSet_descriptor, b) end
 
 function M.ConformanceRequest_new(t) return t or {} end
 


@@ 366,6 368,7 @@ function M.ConformanceRequest_decode(buf)
     return result
 end
 
+function M.ConformanceRequest_decode_lazy(b) return pb.decode_lazy(M.ConformanceRequest_descriptor, b) end
 
 function M.ConformanceResponse_new(t) return t or {} end
 


@@ 573,6 576,7 @@ function M.ConformanceResponse_decode(buf)
     return result
 end
 
+function M.ConformanceResponse_decode_lazy(b) return pb.decode_lazy(M.ConformanceResponse_descriptor, b) end
 
 function M.JspbEncodingConfig_new(t) return t or {} end
 


@@ 618,5 622,6 @@ function M.JspbEncodingConfig_decode(buf)
     return result
 end
 
+function M.JspbEncodingConfig_decode_lazy(b) return pb.decode_lazy(M.JspbEncodingConfig_descriptor, b) end
 
 return M

@@ 177,6 177,7 @@ function M.Result_decode(buf)
     return result
 end
 
+function M.Result_decode_lazy(b) return pb.decode_lazy(M.Result_descriptor, b) end
 
 function M.HelloRequest_new(t) return t or {} end
 


@@ 222,6 223,7 @@ function M.HelloRequest_decode(buf)
     return result
 end
 
+function M.HelloRequest_decode_lazy(b) return pb.decode_lazy(M.HelloRequest_descriptor, b) end
 
 function M.HelloReply_new(t) return t or {} end
 


@@ 267,6 269,7 @@ function M.HelloReply_decode(buf)
     return result
 end
 
+function M.HelloReply_decode_lazy(b) return pb.decode_lazy(M.HelloReply_descriptor, b) end
 
 function M.Event_new(t) return t or {} end
 


@@ 488,6 491,7 @@ function M.Event_decode(buf)
     return result
 end
 
+function M.Event_decode_lazy(b) return pb.decode_lazy(M.Event_descriptor, b) end
 
 function M.Address_new(t) return t or {} end
 


@@ 563,6 567,7 @@ function M.Address_decode(buf)
     return result
 end
 
+function M.Address_decode_lazy(b) return pb.decode_lazy(M.Address_descriptor, b) end
 function M.Address_has_apartment(t) return t.apartment ~= nil end
 function M.Address_clear_apartment(t) t.apartment = nil end
 


@@ 863,6 868,7 @@ function M.Person_decode(buf)
     return result
 end
 
+function M.Person_decode_lazy(b) return pb.decode_lazy(M.Person_descriptor, b) end
 
 -- Service: hello.Greeter
 M.Greeter_service = {

@@ 3439,6 3439,7 @@ function M.TestAllTypesProto3_decode(buf)
     return result
 end
 
+function M.TestAllTypesProto3_decode_lazy(b) return pb.decode_lazy(M.TestAllTypesProto3_descriptor, b) end
 
 function M.TestAllTypesProto3_NestedMessage_new(t) return t or {} end
 


@@ 3500,6 3501,7 @@ function M.TestAllTypesProto3_NestedMessage_decode(buf)
     return result
 end
 
+function M.TestAllTypesProto3_NestedMessage_decode_lazy(b) return pb.decode_lazy(M.TestAllTypesProto3_NestedMessage_descriptor, b) end
 
 function M.ForeignMessage_new(t) return t or {} end
 


@@ 3545,6 3547,7 @@ function M.ForeignMessage_decode(buf)
     return result
 end
 
+function M.ForeignMessage_decode_lazy(b) return pb.decode_lazy(M.ForeignMessage_descriptor, b) end
 
 function M.NullHypothesisProto3_new(t) return t or {} end
 


@@ 3581,6 3584,7 @@ function M.NullHypothesisProto3_decode(buf)
     return result
 end
 
+function M.NullHypothesisProto3_decode_lazy(b) return pb.decode_lazy(M.NullHypothesisProto3_descriptor, b) end
 
 function M.EnumOnlyProto3_new(t) return t or {} end
 


@@ 3617,5 3621,6 @@ function M.EnumOnlyProto3_decode(buf)
     return result
 end
 
+function M.EnumOnlyProto3_decode_lazy(b) return pb.decode_lazy(M.EnumOnlyProto3_descriptor, b) end
 
 return M

@@ 93,21 93,26 @@ pb.finalize_message(M.JspbEncodingConfig_descriptor)
 function M.TestStatus_new(t) return t or {} end
 function M.TestStatus_encode(t) return pb.encode(M.TestStatus_descriptor, t) end
 function M.TestStatus_decode(b) return pb.decode(M.TestStatus_descriptor, b) end
+function M.TestStatus_decode_lazy(b) return pb.decode_lazy(M.TestStatus_descriptor, b) end
 
 function M.FailureSet_new(t) return t or {} end
 function M.FailureSet_encode(t) return pb.encode(M.FailureSet_descriptor, t) end
 function M.FailureSet_decode(b) return pb.decode(M.FailureSet_descriptor, b) end
+function M.FailureSet_decode_lazy(b) return pb.decode_lazy(M.FailureSet_descriptor, b) end
 
 function M.ConformanceRequest_new(t) return t or {} end
 function M.ConformanceRequest_encode(t) return pb.encode(M.ConformanceRequest_descriptor, t) end
 function M.ConformanceRequest_decode(b) return pb.decode(M.ConformanceRequest_descriptor, b) end
+function M.ConformanceRequest_decode_lazy(b) return pb.decode_lazy(M.ConformanceRequest_descriptor, b) end
 
 function M.ConformanceResponse_new(t) return t or {} end
 function M.ConformanceResponse_encode(t) return pb.encode(M.ConformanceResponse_descriptor, t) end
 function M.ConformanceResponse_decode(b) return pb.decode(M.ConformanceResponse_descriptor, b) end
+function M.ConformanceResponse_decode_lazy(b) return pb.decode_lazy(M.ConformanceResponse_descriptor, b) end
 
 function M.JspbEncodingConfig_new(t) return t or {} end
 function M.JspbEncodingConfig_encode(t) return pb.encode(M.JspbEncodingConfig_descriptor, t) end
 function M.JspbEncodingConfig_decode(b) return pb.decode(M.JspbEncodingConfig_descriptor, b) end
+function M.JspbEncodingConfig_decode_lazy(b) return pb.decode_lazy(M.JspbEncodingConfig_descriptor, b) end
 
 return M

@@ 96,28 96,34 @@ pb.finalize_message(M.Person_descriptor)
 function M.Result_new(t) return t or {} end
 function M.Result_encode(t) return pb.encode(M.Result_descriptor, t) end
 function M.Result_decode(b) return pb.decode(M.Result_descriptor, b) end
+function M.Result_decode_lazy(b) return pb.decode_lazy(M.Result_descriptor, b) end
 
 function M.HelloRequest_new(t) return t or {} end
 function M.HelloRequest_encode(t) return pb.encode(M.HelloRequest_descriptor, t) end
 function M.HelloRequest_decode(b) return pb.decode(M.HelloRequest_descriptor, b) end
+function M.HelloRequest_decode_lazy(b) return pb.decode_lazy(M.HelloRequest_descriptor, b) end
 
 function M.HelloReply_new(t) return t or {} end
 function M.HelloReply_encode(t) return pb.encode(M.HelloReply_descriptor, t) end
 function M.HelloReply_decode(b) return pb.decode(M.HelloReply_descriptor, b) end
+function M.HelloReply_decode_lazy(b) return pb.decode_lazy(M.HelloReply_descriptor, b) end
 
 function M.Event_new(t) return t or {} end
 function M.Event_encode(t) return pb.encode(M.Event_descriptor, t) end
 function M.Event_decode(b) return pb.decode(M.Event_descriptor, b) end
+function M.Event_decode_lazy(b) return pb.decode_lazy(M.Event_descriptor, b) end
 
 function M.Address_new(t) return t or {} end
 function M.Address_encode(t) return pb.encode(M.Address_descriptor, t) end
 function M.Address_decode(b) return pb.decode(M.Address_descriptor, b) end
+function M.Address_decode_lazy(b) return pb.decode_lazy(M.Address_descriptor, b) end
 function M.Address_has_apartment(t) return t.apartment ~= nil end
 function M.Address_clear_apartment(t) t.apartment = nil end
 
 function M.Person_new(t) return t or {} end
 function M.Person_encode(t) return pb.encode(M.Person_descriptor, t) end
 function M.Person_decode(b) return pb.decode(M.Person_descriptor, b) end
+function M.Person_decode_lazy(b) return pb.decode_lazy(M.Person_descriptor, b) end
 
 -- Service: hello.Greeter
 M.Greeter_service = {

@@ 237,21 237,26 @@ pb.finalize_message(M.EnumOnlyProto3_descriptor)
 function M.TestAllTypesProto3_new(t) return t or {} end
 function M.TestAllTypesProto3_encode(t) return pb.encode(M.TestAllTypesProto3_descriptor, t) end
 function M.TestAllTypesProto3_decode(b) return pb.decode(M.TestAllTypesProto3_descriptor, b) end
+function M.TestAllTypesProto3_decode_lazy(b) return pb.decode_lazy(M.TestAllTypesProto3_descriptor, b) end
 
 function M.TestAllTypesProto3_NestedMessage_new(t) return t or {} end
 function M.TestAllTypesProto3_NestedMessage_encode(t) return pb.encode(M.TestAllTypesProto3_NestedMessage_descriptor, t) end
 function M.TestAllTypesProto3_NestedMessage_decode(b) return pb.decode(M.TestAllTypesProto3_NestedMessage_descriptor, b) end
+function M.TestAllTypesProto3_NestedMessage_decode_lazy(b) return pb.decode_lazy(M.TestAllTypesProto3_NestedMessage_descriptor, b) end
 
 function M.ForeignMessage_new(t) return t or {} end
 function M.ForeignMessage_encode(t) return pb.encode(M.ForeignMessage_descriptor, t) end
 function M.ForeignMessage_decode(b) return pb.decode(M.ForeignMessage_descriptor, b) end
+function M.ForeignMessage_decode_lazy(b) return pb.decode_lazy(M.ForeignMessage_descriptor, b) end
 
 function M.NullHypothesisProto3_new(t) return t or {} end
 function M.NullHypothesisProto3_encode(t) return pb.encode(M.NullHypothesisProto3_descriptor, t) end
 function M.NullHypothesisProto3_decode(b) return pb.decode(M.NullHypothesisProto3_descriptor, b) end
+function M.NullHypothesisProto3_decode_lazy(b) return pb.decode_lazy(M.NullHypothesisProto3_descriptor, b) end
 
 function M.EnumOnlyProto3_new(t) return t or {} end
 function M.EnumOnlyProto3_encode(t) return pb.encode(M.EnumOnlyProto3_descriptor, t) end
 function M.EnumOnlyProto3_decode(b) return pb.decode(M.EnumOnlyProto3_descriptor, b) end
+function M.EnumOnlyProto3_decode_lazy(b) return pb.decode_lazy(M.EnumOnlyProto3_descriptor, b) end
 
 return M

@@ 411,6 411,14 @@ local function build_writer(f)
     return nil
 end
 
+-- Expose encode_field for callers that need to emit a single field's bytes
+-- without walking a full message (e.g. lazy passthrough re-encode, which
+-- splices original wire segments for untouched fields and calls
+-- encode_field for the dirty ones).
+M.encode_field = function(field, value, out, force)
+    return encode_field(field, value, out, force)
+end
+
 -- compile_writers attaches `f._writer` to each field where the shape is
 -- specialized. Called from pb.finalize_message after the oneof flatten.
 function M.compile_writers(desc)

@@ 15,12 15,19 @@ local grpc    = require('pb.grpc')
 local parser  = require('pb.parser')
 local dynamic = require('pb.dynamic')
 local pbjson  = require('pb.json')
+local lazy    = require('pb.lazy')
 
 return {
     -- High-level codec
     encode = codec.encode,
     decode = codec.decode,
 
+    -- Lazy / zero-copy decode view. See runtime/pb/lazy.lua for the
+    -- :get / :has / :which / :iter / :names surface on the returned
+    -- MessageView (and ArrayView / MapView for repeated and map fields).
+    decode_lazy = lazy.build,
+    lazy        = lazy,
+
     -- Wire-format primitives (exposed for advanced users / tests)
     wire   = wire,
 


@@ 81,9 88,13 @@ return {
     -- Generated code calls this after constructing the fields table so cross-references
     -- (including self-references) can be patched in before sealing.
     finalize_message = function(desc)
-        local fbi = {}
-        for _, f in ipairs(desc.fields) do fbi[f.id] = f end
+        local fbi, fbn = {}, {}
+        for _, f in ipairs(desc.fields) do
+            fbi[f.id] = f
+            fbn[f.name] = f
+        end
         desc.field_by_id = fbi
+        desc.field_by_name = fbn
         -- Pre-compute sibling lists for each oneof field so decode can clear
         -- them in O(k) without rescanning.
         --

@@ 0,0 1,729 @@
+-- Lazy, zero-copy decode view over protobuf wire bytes.
+--
+-- Built once via pb.decode_lazy(desc, bytes). Indexes the wire bytes at
+-- construct time (single pass), then materializes individual fields on
+-- access. Nested messages, repeated fields, and maps are themselves
+-- returned as lazy sub-views; values are decoded on `:get` / `:at`.
+--
+-- The underlying Lua string is kept GC-anchored on the view table so
+-- substring-based passthrough remains valid for the view's lifetime.
+--
+-- Phase 1: read-only API (:get / :has / :which / :iter / :names plus
+-- ArrayView :len/:at/:iter and MapView :get/:has/:keys/:iter). Mutation
+-- and passthrough re-encode arrive in phase 2.
+--
+-- WKT and any descriptor carrying `desc.decode` are eagerly decoded and
+-- wrapped in an EagerView with the same getter surface, so callers don't
+-- have to special-case Timestamp/Duration/Struct/etc.
+
+local wire  = require('pb.wire')
+local codec = require('pb.codec')
+
+local M = {}
+
+-- Forward decls: views can reference each other.
+local build_msg_view
+local build_array_view
+local build_map_view
+
+-- ---------------------------------------------------------------------------
+-- Index build
+--
+-- Single pass over `bytes`. Each tag+value occupies one `segment`:
+--   {id = <field id>,
+--    tag_start = <1-based offset of tag's first byte>,
+--    val_start = <1-based offset of value's first byte>,
+--    next_start = <1-based offset just past this segment>,
+--    wt = <wire type>}
+--
+-- `segments` is the wire-order list (used by iter and by passthrough re-encode
+-- in phase 2). `by_id` maps id -> list-of-segments, even for singular fields
+-- (multiple wire entries for the same singular id are legal per spec: scalar
+-- last-wins, message-merge — both consumers can walk the list).
+-- ---------------------------------------------------------------------------
+local function index_bytes(desc, bytes)
+    local pos, lim = 1, #bytes
+    local segments = {}
+    local by_id = {}
+    local fbi = desc.field_by_id
+
+    while pos <= lim do
+        local tag_start = pos
+        local id, wt, npos = wire.decode_tag(bytes, pos)
+        local val_start = npos
+        local next_start = wire.skip_field(bytes, npos, wt)
+        local seg = {
+            id = id,
+            tag_start = tag_start,
+            val_start = val_start,
+            next_start = next_start,
+            wt = wt,
+        }
+        segments[#segments + 1] = seg
+        if fbi[id] ~= nil then
+            local list = by_id[id]
+            if list == nil then
+                by_id[id] = {seg}
+            else
+                list[#list + 1] = seg
+            end
+        end
+        pos = next_start
+    end
+    return segments, by_id
+end
+
+-- ---------------------------------------------------------------------------
+-- Per-field materialization
+-- ---------------------------------------------------------------------------
+
+-- Decode a single value from one segment, given the singular field shape.
+-- For message fields this returns another lazy view; for scalar/enum it
+-- returns the materialized Lua/cdata value.
+local function read_singular(field, bytes, seg)
+    local kind = field.kind
+    if kind == 'scalar' then
+        local h = wire.TYPE_INFO[field.proto_type]
+        local v = h.decode(bytes, seg.val_start)
+        return v
+    elseif kind == 'enum' then
+        local u = wire.decode_varint(bytes, seg.val_start)
+        return tonumber(u)
+    elseif kind == 'message' then
+        local payload = wire.decode_len(bytes, seg.val_start)
+        if field.message.decode ~= nil then
+            -- WKT (or any custom-decode override). Eager.
+            return field.message.decode(payload)
+        end
+        return build_msg_view(field.message, payload)
+    end
+    error("read_singular: unknown kind " .. tostring(kind), 0)
+end
+
+-- For singular fields, semantics for multiple wire entries with the same id:
+--   - scalar/enum: last wins.
+--   - message: merged. We delegate to the eager codec by concatenating the
+--     per-entry payloads and feeding them to pb.codec.decode, which honors
+--     proto3 merge rules. (Multi-entry singular messages are rare; this is
+--     the off-fast-path correctness branch.)
+local function read_singular_list(field, bytes, list)
+    if field.kind ~= 'message' or #list == 1 then
+        return read_singular(field, bytes, list[#list])
+    end
+    -- Multi-segment message: concatenate the inner payloads and eager-decode.
+    -- This loses the lazy sub-view but is the right thing semantically.
+    local parts = {}
+    for i = 1, #list do
+        local seg = list[i]
+        local payload = wire.decode_len(bytes, seg.val_start)
+        parts[i] = payload
+    end
+    local merged = table.concat(parts)
+    if field.message.decode ~= nil then
+        return field.message.decode(merged)
+    end
+    return build_msg_view(field.message, merged)
+end
+
+-- ---------------------------------------------------------------------------
+-- ArrayView: lazy view over a repeated field.
+--
+-- For unpacked repeated, each wire entry is one element (one segment).
+-- For packed repeated (scalars/enums), one segment contains a length-delimited
+-- payload with all elements; we walk the payload on demand. We never
+-- materialize the full element array unless the user iterates it all.
+-- ---------------------------------------------------------------------------
+
+local ArrayView = {}
+ArrayView.__index = ArrayView
+
+-- Pre-expand packed payloads into a uniform per-element segment list at
+-- construct time. Cheaper to do once than to re-scan on every :at(i).
+local function expand_packed(field, bytes, packed_seg)
+    local h = field.kind == 'scalar' and wire.TYPE_INFO[field.proto_type] or nil
+    -- Decode the length prefix to know where elements live.
+    local payload_start = packed_seg.val_start
+    local b = bytes:byte(payload_start)
+    local payload_len, hdr_end
+    if b < 0x80 then
+        payload_len = b
+        hdr_end = payload_start + 1
+    else
+        local v, npos = wire.decode_varint(bytes, payload_start)
+        payload_len = tonumber(v)
+        hdr_end = npos
+    end
+    local lim = hdr_end + payload_len
+    local elems = {}
+    local p = hdr_end
+    while p < lim do
+        local v_start = p
+        local v_next
+        if field.kind == 'scalar' then
+            -- Skip according to the scalar's actual wire type.
+            v_next = wire.skip_field(bytes, p, h.wire)
+        else
+            -- Packed enums: varint per element.
+            v_next = wire.skip_field(bytes, p, wire.WIRE_VARINT)
+        end
+        elems[#elems + 1] = {val_start = v_start, val_next = v_next}
+        p = v_next
+    end
+    return elems
+end
+
+local function build_array_view_impl(field, bytes, segs)
+    -- For packed payloads (single segment, WIRE_LEN, but scalar field uses
+    -- a non-LEN wire type), elements live inside that one segment.
+    -- For unpacked, each segment is one element.
+    local mode  -- 'packed' | 'unpacked'
+    local elements  -- list of {val_start = ...} entries
+    if field.kind == 'scalar' then
+        local h = wire.TYPE_INFO[field.proto_type]
+        if h.wire ~= wire.WIRE_LEN and #segs == 1 and segs[1].wt == wire.WIRE_LEN then
+            mode = 'packed'
+            elements = expand_packed(field, bytes, segs[1])
+        else
+            mode = 'unpacked'
+            elements = segs
+        end
+    elseif field.kind == 'enum' then
+        if #segs == 1 and segs[1].wt == wire.WIRE_LEN then
+            mode = 'packed'
+            elements = expand_packed(field, bytes, segs[1])
+        else
+            mode = 'unpacked'
+            elements = segs
+        end
+    else
+        -- repeated message: always one segment per element, never packed.
+        mode = 'unpacked'
+        elements = segs
+    end
+    return setmetatable({
+        _field = field,
+        _bytes = bytes,
+        _mode  = mode,
+        _elements = elements,
+    }, ArrayView)
+end
+
+build_array_view = build_array_view_impl
+
+function ArrayView:len()
+    return #self._elements
+end
+
+function ArrayView:at(i)
+    local entry = self._elements[i]
+    if entry == nil then return nil end
+    local field, bytes = self._field, self._bytes
+    if self._mode == 'packed' then
+        if field.kind == 'enum' then
+            local u = wire.decode_varint(bytes, entry.val_start)
+            return tonumber(u)
+        end
+        local h = wire.TYPE_INFO[field.proto_type]
+        return (h.decode(bytes, entry.val_start))
+    end
+    return read_singular(field, bytes, entry)
+end
+
+function ArrayView:iter()
+    local view, i = self, 0
+    local n = #self._elements
+    return function()
+        i = i + 1
+        if i > n then return nil end
+        return i, view:at(i)
+    end
+end
+
+function ArrayView:tolist()
+    local out = {}
+    for i = 1, #self._elements do out[i] = self:at(i) end
+    return out
+end
+
+-- ---------------------------------------------------------------------------
+-- MapView: lazy view over a map<K,V> field.
+--
+-- Each wire entry is one length-delimited submessage with two fields:
+--   id=1 (key), id=2 (value). The submessage may omit either when the
+-- value equals its proto3 default.
+--
+-- For O(1) `:get(k)`, we decode keys lazily but cache the key->entry map on
+-- first key-lookup or first :iter. Until then, only the per-entry payload
+-- offsets are known.
+-- ---------------------------------------------------------------------------
+
+local MapView = {}
+MapView.__index = MapView
+
+local function decode_map_entry(field, bytes, seg)
+    local key_field, val_field = field.key, field.value
+    -- Read the outer LEN to find the entry payload.
+    local b = bytes:byte(seg.val_start)
+    local payload_len, hdr_end
+    if b < 0x80 then
+        payload_len = b; hdr_end = seg.val_start + 1
+    else
+        local v, npos = wire.decode_varint(bytes, seg.val_start)
+        payload_len = tonumber(v); hdr_end = npos
+    end
+    local lim = hdr_end + payload_len
+    local key, val
+    local p = hdr_end
+    while p < lim do
+        local eid, ewt, np = wire.decode_tag(bytes, p)
+        p = np
+        if eid == 1 then
+            if key_field.kind == 'scalar' then
+                key, p = wire.TYPE_INFO[key_field.proto_type].decode(bytes, p)
+            elseif key_field.kind == 'enum' then
+                local u; u, p = wire.decode_varint(bytes, p); key = tonumber(u)
+            end
+        elseif eid == 2 then
+            if val_field.kind == 'scalar' then
+                val, p = wire.TYPE_INFO[val_field.proto_type].decode(bytes, p)
+            elseif val_field.kind == 'enum' then
+                local u; u, p = wire.decode_varint(bytes, p); val = tonumber(u)
+            elseif val_field.kind == 'message' then
+                local payload; payload, p = wire.decode_len(bytes, p)
+                if val_field.message.decode ~= nil then
+                    val = val_field.message.decode(payload)
+                else
+                    val = build_msg_view(val_field.message, payload)
+                end
+            end
+        else
+            p = wire.skip_field(bytes, p, ewt)
+        end
+    end
+    if key == nil then
+        if key_field.kind == 'scalar' then
+            local pt = key_field.proto_type
+            if pt == 'string' or pt == 'bytes' then key = ''
+            elseif pt == 'bool' then key = false
+            else key = 0 end
+        else key = 0 end
+    end
+    if val == nil then
+        if val_field.kind == 'scalar' then
+            local pt = val_field.proto_type
+            if pt == 'string' or pt == 'bytes' then val = ''
+            elseif pt == 'bool' then val = false
+            else val = 0 end
+        elseif val_field.kind == 'enum' then val = 0
+        elseif val_field.kind == 'message' then val = {} end
+    end
+    return key, val
+end
+
+local function build_map_view_impl(field, bytes, segs)
+    return setmetatable({
+        _field = field,
+        _bytes = bytes,
+        _segs  = segs,
+        -- _by_key populated lazily on first :get/:has/:iter call.
+    }, MapView)
+end
+
+build_map_view = build_map_view_impl
+
+local function map_ensure_index(self)
+    if self._by_key ~= nil then return end
+    local field, bytes, segs = self._field, self._bytes, self._segs
+    local by_key = {}
+    local keys = {}
+    for i = 1, #segs do
+        local k, v = decode_map_entry(field, bytes, segs[i])
+        if by_key[k] == nil then keys[#keys + 1] = k end
+        by_key[k] = v  -- duplicate keys: last wins (matches eager decode)
+    end
+    self._by_key = by_key
+    self._keys = keys
+end
+
+function MapView:get(k)
+    map_ensure_index(self)
+    return self._by_key[k]
+end
+
+function MapView:has(k)
+    map_ensure_index(self)
+    return self._by_key[k] ~= nil
+end
+
+function MapView:keys()
+    map_ensure_index(self)
+    local out = {}
+    for i = 1, #self._keys do out[i] = self._keys[i] end
+    return out
+end
+
+function MapView:iter()
+    map_ensure_index(self)
+    local keys, by_key = self._keys, self._by_key
+    local i = 0
+    return function()
+        i = i + 1
+        local k = keys[i]
+        if k == nil then return nil end
+        return k, by_key[k]
+    end
+end
+
+function MapView:totable()
+    map_ensure_index(self)
+    local out = {}
+    for k, v in pairs(self._by_key) do out[k] = v end
+    return out
+end
+
+-- ---------------------------------------------------------------------------
+-- MessageView (top-level)
+-- ---------------------------------------------------------------------------
+
+local MessageView = {}
+MessageView.__index = MessageView
+
+local function build_msg_view_impl(desc, bytes)
+    if desc.decode ~= nil then
+        -- WKT / custom-decode descriptor. Eager-wrap so the API stays uniform.
+        local materialized = desc.decode(bytes)
+        return setmetatable({
+            _desc = desc,
+            _eager = materialized,
+            _eager_only = true,
+        }, MessageView)
+    end
+    local segments, by_id = index_bytes(desc, bytes)
+    return setmetatable({
+        _desc = desc,
+        _bytes = bytes,
+        _segments = segments,
+        _by_id = by_id,
+        _cache = {},
+        -- Parallel array of cached sub-MessageViews so :is_dirty can
+        -- ipairs over it instead of pairs(_cache) — `pairs` over a hash
+        -- compiles to bytecode ISNEXT, NYI in Tarantool's LuaJIT 2.1.
+        _sub_msg_views = {},
+    }, MessageView)
+end
+
+build_msg_view = build_msg_view_impl
+M.build = build_msg_view_impl
+
+-- :get(name) -> decoded value, or nil if not on wire (and no _cache entry).
+function MessageView:get(name)
+    if self._eager_only then return self._eager[name] end
+    local cache = self._cache
+    local v = cache[name]
+    if v ~= nil then return v end
+    local field = self._desc.field_by_name[name]
+    if field == nil then return nil end
+    local segs = self._by_id[field.id]
+    if segs == nil then return nil end
+    if field.kind == 'map' then
+        v = build_map_view(field, self._bytes, segs)
+    elseif field.repeated then
+        v = build_array_view(field, self._bytes, segs)
+    else
+        v = read_singular_list(field, self._bytes, segs)
+    end
+    cache[name] = v
+    -- Track sub-MessageViews on a flat array so :is_dirty can walk
+    -- with ipairs (see _sub_msg_views note above).
+    if type(v) == 'table' and getmetatable(v) == MessageView then
+        local s = self._sub_msg_views
+        s[#s + 1] = v
+    end
+    return v
+end
+
+-- :has(name) -> was this field present on the wire?
+function MessageView:has(name)
+    if self._eager_only then return self._eager[name] ~= nil end
+    local field = self._desc.field_by_name[name]
+    if field == nil then return false end
+    return self._by_id[field.id] ~= nil
+end
+
+-- :which(oneof_name) -> name of the active branch, or nil.
+-- Proto3 last-wins: if multiple branches appeared on the wire, the one
+-- whose final segment came last in wire order is active.
+function MessageView:which(oneof_name)
+    if self._eager_only then
+        local oneofs = self._desc.oneofs
+        if oneofs == nil then return nil end
+        local members = oneofs[oneof_name]
+        if members == nil then return nil end
+        for i = 1, #members do
+            if self._eager[members[i]] ~= nil then return members[i] end
+        end
+        return nil
+    end
+    local oneofs = self._desc.oneofs
+    if oneofs == nil then return nil end
+    local members = oneofs[oneof_name]
+    if members == nil then return nil end
+    local member_set = {}
+    for i = 1, #members do member_set[members[i]] = true end
+    local fbn = self._desc.field_by_name
+    local active, active_pos
+    local segs = self._segments
+    for i = 1, #segs do
+        local f = self._desc.field_by_id[segs[i].id]
+        if f ~= nil and member_set[f.name] then
+            if active_pos == nil or segs[i].tag_start > active_pos then
+                active = f.name
+                active_pos = segs[i].tag_start
+            end
+        end
+    end
+    return active
+end
+
+-- :names() -> iterator yielding present field names in wire order
+-- (deduplicated; each field appears once even when it has multiple wire entries).
+function MessageView:names()
+    if self._eager_only then
+        local fields, i = self._desc.fields, 0
+        local eager = self._eager
+        return function()
+            while true do
+                i = i + 1
+                local f = fields[i]
+                if f == nil then return nil end
+                if eager[f.name] ~= nil then return f.name end
+            end
+        end
+    end
+    local segments = self._segments
+    local fbi = self._desc.field_by_id
+    local emitted = {}
+    local i = 0
+    return function()
+        while true do
+            i = i + 1
+            local seg = segments[i]
+            if seg == nil then return nil end
+            local f = fbi[seg.id]
+            if f ~= nil and emitted[f.name] == nil then
+                emitted[f.name] = true
+                return f.name
+            end
+        end
+    end
+end
+
+-- :iter() -> iterator yielding (name, value) for present fields,
+-- decoding each value on demand. Order = wire order, deduplicated.
+function MessageView:iter()
+    if self._eager_only then
+        local fields, i = self._desc.fields, 0
+        local eager = self._eager
+        return function()
+            while true do
+                i = i + 1
+                local f = fields[i]
+                if f == nil then return nil end
+                local v = eager[f.name]
+                if v ~= nil then return f.name, v end
+            end
+        end
+    end
+    local segments = self._segments
+    local fbi = self._desc.field_by_id
+    local emitted = {}
+    local view = self
+    local i = 0
+    return function()
+        while true do
+            i = i + 1
+            local seg = segments[i]
+            if seg == nil then return nil end
+            local f = fbi[seg.id]
+            if f ~= nil and emitted[f.name] == nil then
+                emitted[f.name] = true
+                return f.name, view:get(f.name)
+            end
+        end
+    end
+end
+
+-- ---------------------------------------------------------------------------
+-- Mutation
+-- ---------------------------------------------------------------------------
+
+-- :set(name, value) marks a field dirty. Subsequent :encode() emits the
+-- new value via the codec; other fields passthrough their original bytes.
+-- The provided value can be any shape the eager encoder accepts (Lua
+-- table/string/number/cdata/array). Passing a MessageView as `value` is
+-- supported but materializes it via :totable() on encode.
+function MessageView:set(name, value)
+    if self._eager_only then
+        self._eager[name] = value
+        self._eager_dirty = true
+        return
+    end
+    local field = self._desc.field_by_name[name]
+    if field == nil then
+        error("unknown field '" .. tostring(name) .. "' on " .. self._desc.name, 0)
+    end
+    if self._dirty == nil then self._dirty = {} end
+    self._dirty[name] = true
+    self._cache[name] = value
+end
+
+-- :is_dirty() returns true if this view has had :set called, OR if any
+-- cached sub-view (e.g. a nested MessageView accessed via :get) has been
+-- mutated. Used by parent :encode to decide whether to splice or re-encode.
+--
+-- We walk _sub_msg_views (a flat array of cached sub-MessageViews) with
+-- ipairs instead of pairs(_cache). Hash-keyed pairs compiles to ISNEXT
+-- which is NYI in Tarantool's LuaJIT 2.1 — keeping is_dirty on an array
+-- lets the encode-time JIT trace stay attached. Array fields and map
+-- fields aren't recursed into: phase 2 mutation is on MessageView only.
+function MessageView:is_dirty()
+    if self._eager_only then return self._eager_dirty == true end
+    local d = self._dirty
+    if d ~= nil and next(d) ~= nil then return true end
+    local subs = self._sub_msg_views
+    for i = 1, #subs do
+        if subs[i]:is_dirty() then return true end
+    end
+    return false
+end
+
+-- Materialize a possibly-view value so encode_field can consume it.
+-- Lazy sub-views are converted to plain tables; raw values pass through.
+local function materialize(value)
+    if type(value) ~= 'table' then return value end
+    local m = getmetatable(value)
+    if m == MessageView then
+        return value:totable()
+    elseif m == ArrayView then
+        return value:tolist()
+    elseif m == MapView then
+        return value:totable()
+    end
+    return value
+end
+
+-- Recursively materialize a value to a plain Lua structure suitable for
+-- the eager codec. For a MessageView, we walk every present field —
+-- this is the cost of mutation; pure-read messages stay lazy.
+function MessageView:totable()
+    if self._eager_only then return self._eager end
+    local out = {}
+    for name in self:names() do
+        local v = self:get(name)
+        out[name] = materialize(v)
+    end
+    -- Preserve unknown fields for round-trip.
+    if self._segments then
+        local unknown = {}
+        local fbi = self._desc.field_by_id
+        for i = 1, #self._segments do
+            local seg = self._segments[i]
+            if fbi[seg.id] == nil then
+                unknown[#unknown + 1] = self._bytes:sub(seg.tag_start, seg.next_start - 1)
+            end
+        end
+        if #unknown > 0 then out._unknown_fields = table.concat(unknown) end
+    end
+    return out
+end
+
+-- :encode() emits bytes. Three modes:
+--   1. WKT eager-wrap: delegate to desc.encode on the materialized table.
+--   2. Untouched (no dirty fields, no dirty sub-views): emit the original
+--      bytes verbatim — perfect byte-for-byte passthrough.
+--   3. Mixed: walk fields in id order, splice clean segments, encode dirty
+--      values fresh. Unknown segments are emitted at the end.
+function MessageView:encode()
+    if self._eager_only then
+        return self._desc.encode(self._eager)
+    end
+    if not self:is_dirty() then
+        return self._bytes
+    end
+
+    local out = {}
+    local fields = self._desc.fields
+    local bytes = self._bytes
+    local by_id = self._by_id
+    local dirty = self._dirty or {}
+    local cache = self._cache
+
+    -- Active-oneof resolution mirrors codec.encode_message:
+    -- a oneof field is only emitted if it's the active branch.
+    local active
+    local oneofs_list = self._desc.oneofs_list
+    if oneofs_list then
+        active = {}
+        for i = 1, #oneofs_list do
+            local oo = oneofs_list[i]
+            local members = oo.members
+            for j = 1, #members do
+                local fname = members[j]
+                if dirty[fname] or by_id[self._desc.field_by_name[fname].id] then
+                    active[oo.name] = fname
+                end
+            end
+        end
+    end
+
+    for i = 1, #fields do
+        local f = fields[i]
+        local fname = f.name
+        local is_dirty = dirty[fname] == true
+        local is_sub_dirty = false
+        if not is_dirty then
+            local cached = cache[fname]
+            if cached ~= nil and type(cached) == 'table'
+                and getmetatable(cached) == MessageView and cached:is_dirty() then
+                is_sub_dirty = true
+            end
+        end
+
+        if f.oneof and active and active[f.oneof] ~= fname then
+            -- Inactive oneof branch: skip entirely.
+        elseif is_dirty or is_sub_dirty then
+            local v = cache[fname]
+            v = materialize(v)
+            codec.encode_field(f, v, out, f.optional or (f.oneof ~= nil))
+        elseif by_id[f.id] then
+            local segs = by_id[f.id]
+            for j = 1, #segs do
+                local seg = segs[j]
+                out[#out + 1] = bytes:sub(seg.tag_start, seg.next_start - 1)
+            end
+        end
+    end
+
+    -- Unknown segments preserved at the end (matches codec's _unknown_fields
+    -- trailer convention).
+    local fbi = self._desc.field_by_id
+    local segments = self._segments
+    for i = 1, #segments do
+        local seg = segments[i]
+        if fbi[seg.id] == nil then
+            out[#out + 1] = bytes:sub(seg.tag_start, seg.next_start - 1)
+        end
+    end
+
+    return table.concat(out)
+end
+
+-- ---------------------------------------------------------------------------
+-- Public entry
+-- ---------------------------------------------------------------------------
+
+M.MessageView = MessageView
+M.ArrayView   = ArrayView
+M.MapView     = MapView
+
+return M

@@ 86,3 86,24 @@ for _, mode in ipairs({'full', 'runtime'}) do
         end
     end
 end
+
+-- Lazy passthrough: decode_lazy then :encode() must produce bytes
+-- byte-identical to the golden. This is the strongest claim for the
+-- read-only zero-copy path — untouched views skip re-emission entirely.
+for _, mode in ipairs({'full', 'runtime'}) do
+    local hello = require(mode .. '.hello.hello_pb')
+    local g = t.group('interop_lazy.' .. mode)
+
+    for _, fx in ipairs(FIXTURES) do
+        g['test_' .. fx.name] = function()
+            local golden = slurp(fx.bin_path)
+            local short = fx.full_name:gsub('^hello%.', '')
+            local decode_lazy_fn = hello[short .. '_decode_lazy']
+            t.assert(decode_lazy_fn, 'no decode_lazy for ' .. fx.full_name)
+
+            local v = decode_lazy_fn(golden)
+            t.assert_equals(hex(v:encode()), hex(golden),
+                ('lazy passthrough mismatch on %s'):format(fx.name))
+        end
+    end
+end

@@ 0,0 1,328 @@
+-- Tests for the lazy / zero-copy decode view (runtime/pb/lazy.lua).
+-- Parameterized over both codegen modes — the generated _decode_lazy
+-- shim is identical in both, but the descriptor shape it consumes
+-- comes from each mode's emitted module.
+
+local t = require('luatest')
+local ffi = require('ffi')
+local pb = require('pb')
+
+local MODES = {'full', 'runtime'}
+
+for _, mode in ipairs(MODES) do
+    local g = t.group('lazy.' .. mode)
+    local hello = require(mode .. '.hello.hello_pb')
+
+    -- ---- Basic scalar access ----
+
+    g.test_singular_scalars_decode_on_demand = function()
+        local enc = hello.Address_encode({street = 'Main St', city = 'SF', zip = 42})
+        local v = hello.Address_decode_lazy(enc)
+        t.assert_equals(v:get('street'), 'Main St')
+        t.assert_equals(v:get('city'), 'SF')
+        t.assert_equals(v:get('zip'), 42)
+        t.assert_equals(v:get('apartment'), nil, 'absent optional')
+    end
+
+    g.test_has_reports_wire_presence = function()
+        local enc = hello.Address_encode({street = 'X'})
+        local v = hello.Address_decode_lazy(enc)
+        t.assert_equals(v:has('street'), true)
+        t.assert_equals(v:has('city'), false)
+        t.assert_equals(v:has('zip'), false)
+    end
+
+    g.test_get_caches_repeated_calls = function()
+        local enc = hello.Address_encode({street = 'Main'})
+        local v = hello.Address_decode_lazy(enc)
+        local a = v:get('street')
+        local b = v:get('street')
+        t.assert_is(a, b, 'string values are interned but cache should hit')
+    end
+
+    -- ---- Repeated fields ----
+
+    g.test_unpacked_repeated_string = function()
+        local enc = hello.Person_encode({emails = {'a@x', 'b@x', 'c@x'}})
+        local v = hello.Person_decode_lazy(enc)
+        local arr = v:get('emails')
+        t.assert_not_equals(arr, nil)
+        t.assert_equals(arr:len(), 3)
+        t.assert_equals(arr:at(1), 'a@x')
+        t.assert_equals(arr:at(2), 'b@x')
+        t.assert_equals(arr:at(3), 'c@x')
+        local seen = {}
+        for i, s in arr:iter() do seen[i] = s end
+        t.assert_equals(seen, {'a@x', 'b@x', 'c@x'})
+    end
+
+    g.test_packed_repeated_int32 = function()
+        local enc = hello.Person_encode({lucky_numbers = {7, 13, 42}})
+        local v = hello.Person_decode_lazy(enc)
+        local arr = v:get('lucky_numbers')
+        t.assert_equals(arr:len(), 3)
+        t.assert_equals(arr:at(1), 7)
+        t.assert_equals(arr:at(2), 13)
+        t.assert_equals(arr:at(3), 42)
+    end
+
+    g.test_repeated_message_returns_subviews = function()
+        local enc = hello.Person_encode({
+            friends = {
+                {name = 'Bob', age = 20},
+                {name = 'Carol', age = 30},
+            },
+        })
+        local v = hello.Person_decode_lazy(enc)
+        local fr = v:get('friends')
+        t.assert_equals(fr:len(), 2)
+        t.assert_equals(fr:at(1):get('name'), 'Bob')
+        t.assert_equals(fr:at(1):get('age'), 20)
+        t.assert_equals(fr:at(2):get('name'), 'Carol')
+    end
+
+    g.test_absent_repeated_is_nil = function()
+        local enc = hello.Person_encode({name = 'X'})
+        local v = hello.Person_decode_lazy(enc)
+        t.assert_equals(v:get('emails'), nil)
+        t.assert_equals(v:has('emails'), false)
+    end
+
+    -- ---- Nested singular message ----
+
+    g.test_nested_message_subview = function()
+        local enc = hello.Person_encode({
+            name = 'Alice',
+            address = {street = 'Main', city = 'Springfield'},
+        })
+        local v = hello.Person_decode_lazy(enc)
+        local addr = v:get('address')
+        t.assert_equals(addr:get('street'), 'Main')
+        t.assert_equals(addr:get('city'), 'Springfield')
+    end
+
+    -- ---- Map fields ----
+
+    g.test_map_get_and_has = function()
+        local enc = hello.Person_encode({ages_by_nickname = {alice = 30, bob = 25}})
+        local v = hello.Person_decode_lazy(enc)
+        local m = v:get('ages_by_nickname')
+        t.assert_equals(m:get('alice'), 30)
+        t.assert_equals(m:get('bob'), 25)
+        t.assert_equals(m:has('alice'), true)
+        t.assert_equals(m:has('zzz'), false)
+        t.assert_equals(m:get('zzz'), nil)
+    end
+
+    g.test_map_keys_and_iter = function()
+        local enc = hello.Person_encode({ages_by_nickname = {alice = 30}})
+        local v = hello.Person_decode_lazy(enc)
+        local m = v:get('ages_by_nickname')
+        t.assert_equals(m:keys(), {'alice'})
+        local seen = {}
+        for k, val in m:iter() do seen[k] = val end
+        t.assert_equals(seen, {alice = 30})
+    end
+
+    g.test_map_message_values_are_subviews = function()
+        local enc = hello.Person_encode({
+            addresses_by_label = {home = {street = 'Main', city = 'SF'}},
+        })
+        local v = hello.Person_decode_lazy(enc)
+        local m = v:get('addresses_by_label')
+        local home = m:get('home')
+        t.assert_equals(home:get('street'), 'Main')
+        t.assert_equals(home:get('city'), 'SF')
+    end
+
+    -- ---- Oneof ----
+
+    g.test_oneof_which_text_branch = function()
+        local enc = hello.Result_encode({id = 1, text = 'ok'})
+        local v = hello.Result_decode_lazy(enc)
+        t.assert_equals(v:which('outcome'), 'text')
+        t.assert_equals(v:get('text'), 'ok')
+        -- Inactive branches: not on the wire, so :has is false.
+        t.assert_equals(v:has('code'), false)
+        t.assert_equals(v:has('details'), false)
+    end
+
+    g.test_oneof_which_message_branch = function()
+        local enc = hello.Result_encode({details = {street = 'Main'}})
+        local v = hello.Result_decode_lazy(enc)
+        t.assert_equals(v:which('outcome'), 'details')
+        local d = v:get('details')
+        t.assert_equals(d:get('street'), 'Main')
+    end
+
+    g.test_oneof_no_branch_set = function()
+        local enc = hello.Result_encode({id = 1})
+        local v = hello.Result_decode_lazy(enc)
+        t.assert_equals(v:which('outcome'), nil)
+    end
+
+    -- ---- iter / names ----
+
+    g.test_iter_yields_present_fields_in_wire_order = function()
+        local enc = hello.Address_encode({street = 'A', city = 'B', zip = 7})
+        local v = hello.Address_decode_lazy(enc)
+        local seen = {}
+        for name, val in v:iter() do seen[#seen + 1] = {name, val} end
+        t.assert_equals(seen, {{'street', 'A'}, {'city', 'B'}, {'zip', 7}})
+    end
+
+    g.test_names_yields_present_fields_only = function()
+        local enc = hello.Address_encode({street = 'A', zip = 7})
+        local v = hello.Address_decode_lazy(enc)
+        local names = {}
+        for name in v:names() do names[#names + 1] = name end
+        t.assert_equals(names, {'street', 'zip'})
+    end
+
+    g.test_iter_skips_unknown_fields = function()
+        -- Hand-craft bytes with an extra unknown field id.
+        local known = hello.Address_encode({street = 'X'})
+        local extra = string.char(0x68, 0x05)  -- tag (id=13, varint), value=5
+        local v = hello.Address_decode_lazy(known .. extra)
+        local names = {}
+        for name in v:names() do names[#names + 1] = name end
+        t.assert_equals(names, {'street'})
+        t.assert_equals(v:get('street'), 'X')
+    end
+
+    -- ---- WKT eager-wrap ----
+
+    g.test_wkt_timestamp_eager_wrapped = function()
+        local datetime = require('datetime')
+        local dt = datetime.new({timestamp = 1700000000, nsec = 0})
+        local enc = hello.Event_encode({title = 'launch', created_at = dt})
+        local v = hello.Event_decode_lazy(enc)
+        t.assert_equals(v:get('title'), 'launch')
+        local ts = v:get('created_at')
+        -- WKT descriptors carry desc.decode; lazy delegates to it,
+        -- producing whatever the eager codec produces — for Timestamp,
+        -- a datetime cdata equal to the original.
+        t.assert_equals(ts, dt)
+    end
+
+    -- ---- 64-bit cdata correctness ----
+
+    g.test_fixed64_uint64_cdata = function()
+        local big = ffi.cast('uint64_t', 0xdeadbeefcafebabeULL)
+        local enc = hello.Person_encode({user_id = big})
+        local v = hello.Person_decode_lazy(enc)
+        local got = v:get('user_id')
+        t.assert_equals(ffi.cast('uint64_t', got), big)
+    end
+
+    -- ---- Phase 2: mutation + passthrough re-encode ----
+
+    g.test_untouched_view_round_trips_bytes_verbatim = function()
+        local orig = hello.Person_encode({
+            name = 'Alice', age = 30,
+            emails = {'a@x', 'b@x'},
+            address = {street = 'Main', city = 'SF', zip = 100},
+        })
+        local v = hello.Person_decode_lazy(orig)
+        t.assert_equals(v:encode(), orig, 'untouched lazy view -> identical bytes')
+    end
+
+    g.test_set_singular_scalar_round_trips_via_eager = function()
+        local orig = hello.Address_encode({street = 'A', city = 'B', zip = 1})
+        local v = hello.Address_decode_lazy(orig)
+        v:set('city', 'C')
+        local out = v:encode()
+        local eager = hello.Address_decode(out)
+        t.assert_equals(eager.street, 'A')
+        t.assert_equals(eager.city, 'C')
+        t.assert_equals(eager.zip, 1)
+    end
+
+    g.test_set_repeated_replaces_entire_field = function()
+        local orig = hello.Person_encode({emails = {'a@x', 'b@x'}})
+        local v = hello.Person_decode_lazy(orig)
+        v:set('emails', {'new@x'})
+        local eager = hello.Person_decode(v:encode())
+        t.assert_equals(eager.emails, {'new@x'})
+    end
+
+    g.test_set_singular_message_passthrough_for_others = function()
+        local orig = hello.Person_encode({
+            name = 'Alice', age = 30,
+            emails = {'a@x'},
+            address = {street = 'Old', city = 'X'},
+        })
+        local v = hello.Person_decode_lazy(orig)
+        v:set('address', {street = 'New', city = 'Y'})
+        local eager = hello.Person_decode(v:encode())
+        t.assert_equals(eager.name, 'Alice')
+        t.assert_equals(eager.age, 30)
+        t.assert_equals(eager.emails, {'a@x'})
+        t.assert_equals(eager.address.street, 'New')
+        t.assert_equals(eager.address.city, 'Y')
+    end
+
+    g.test_unknown_fields_preserved_through_set = function()
+        local known = hello.Address_encode({street = 'X'})
+        local extra = string.char(0x68, 0x05)  -- id=13, varint, value=5
+        local v = hello.Address_decode_lazy(known .. extra)
+        v:set('zip', 99)
+        local out = v:encode()
+        -- Unknown field bytes should still be present in the output.
+        t.assert(out:find(extra, 1, true) ~= nil, 'unknown bytes preserved')
+        local eager = hello.Address_decode(out)
+        t.assert_equals(eager.street, 'X')
+        t.assert_equals(eager.zip, 99)
+    end
+
+    g.test_oneof_set_clears_other_branches = function()
+        local orig = hello.Result_encode({id = 1, text = 'hello'})
+        local v = hello.Result_decode_lazy(orig)
+        v:set('code', 42)
+        v:set('text', nil)  -- explicit clear
+        local eager = hello.Result_decode(v:encode())
+        t.assert_equals(eager.id, 1)
+        t.assert_equals(eager.code, 42)
+        t.assert_equals(eager.text, nil)
+    end
+
+    g.test_sub_view_mutation_propagates_to_parent_encode = function()
+        local orig = hello.Person_encode({
+            name = 'Alice',
+            address = {street = 'Old', city = 'X', zip = 1},
+        })
+        local v = hello.Person_decode_lazy(orig)
+        local addr = v:get('address')
+        addr:set('street', 'New')
+        local eager = hello.Person_decode(v:encode())
+        t.assert_equals(eager.name, 'Alice')
+        t.assert_equals(eager.address.street, 'New')
+        t.assert_equals(eager.address.city, 'X')
+    end
+
+    -- ---- :get matches eager :decode ----
+
+    g.test_lazy_get_matches_eager_decode = function()
+        local p = {
+            name    = 'Alice',
+            age     = 30,
+            emails  = {'a@x', 'b@x'},
+            status  = hello.Status.OK,
+            address = {street = 'Main', city = 'SF', zip = 100},
+            lucky_numbers = {1, 2, 3},
+            ages_by_nickname = {alice = 30},
+        }
+        local enc = hello.Person_encode(p)
+        local eager = hello.Person_decode(enc)
+        local v = hello.Person_decode_lazy(enc)
+        t.assert_equals(v:get('name'), eager.name)
+        t.assert_equals(v:get('age'), eager.age)
+        t.assert_equals(v:get('status'), eager.status)
+        t.assert_equals(v:get('emails'):tolist(), eager.emails)
+        t.assert_equals(v:get('lucky_numbers'):tolist(), eager.lucky_numbers)
+        t.assert_equals(v:get('ages_by_nickname'):totable(), eager.ages_by_nickname)
+        local addr = v:get('address')
+        t.assert_equals(addr:get('street'), eager.address.street)
+        t.assert_equals(addr:get('zip'), eager.address.zip)
+    end
+end