@@ -63,6 +63,60 @@
  end
  @doc """
  Generate a packfile by streaming bytes through a write callback.
  `write_fn` is invoked with each piece of the pack as it's produced:
  first the 12-byte pack header, then one zlib-compressed entry per
  object, then the trailing 20-byte SHA-1 checksum. SHA-1 is computed
  incrementally so no intermediate full-pack binary is held in memory —
  peak heap is bounded by the largest single object.
  Two arities are exposed:
    * `generate_stream/2` — `write_fn` is `(binary -> any)`, side-effecting.
      Convenient for tests / simple sinks.
    * `generate_stream/3` — `write_fn` is `(binary, acc -> acc)`, threading
      caller state through each chunk. Used by the protocol layer so a
      sideband chunker can accumulate buffered bytes between calls.
  Returns `{pack_sha, count}` for `/2`, `{pack_sha, count, final_acc}` for `/3`.
  """
  @spec generate_stream([object_entry()], (binary() -> any())) ::
          {String.t(), non_neg_integer()}
  def generate_stream(objects, write_fn) when is_list(objects) and is_function(write_fn, 1) do
    {sha, count, nil} =
      generate_stream(objects, nil, fn bytes, _acc ->
        write_fn.(bytes)
        nil
      end)
    {sha, count}
  end
  @spec generate_stream([object_entry()], acc, (binary(), acc -> acc)) ::
          {String.t(), non_neg_integer(), acc}
        when acc: var
  def generate_stream(objects, acc, write_fn)
      when is_list(objects) and is_function(write_fn, 2) do
    count = length(objects)
    header = <<@pack_signature, @pack_version::unsigned-big-32, count::unsigned-big-32>>
    hasher = :crypto.hash_init(:sha)
    acc = write_fn.(header, acc)
    hasher = :crypto.hash_update(hasher, header)
    {hasher, acc} =
      Enum.reduce(objects, {hasher, acc}, fn {type, content, _sha}, {h, a} ->
        entry = IO.iodata_to_binary(encode_entry(type_to_num(type), content))
        a = write_fn.(entry, a)
        {:crypto.hash_update(h, entry), a}
      end)
    checksum = :crypto.hash_final(hasher)
    acc = write_fn.(checksum, acc)
    {Base.encode16(checksum, case: :lower), count, acc}
  end
  @doc """
  Generate both a packfile and its .idx v2 index.
  Returns `{pack_data, idx_data, pack_sha}`.
  """

@@ -164,6 +164,24 @@
    end)
  end
  @doc """
  Maximum payload bytes per sideband-1 frame (pkt-line max 65520 minus
  4-byte length prefix and 1-byte band marker).
  """
  @spec max_sideband_data() :: pos_integer()
  def max_sideband_data, do: @max_sideband_data
  @doc """
  Encode a single sideband frame (one pkt-line). Caller is responsible
  for chunking — `chunk` must be ≤ `max_sideband_data/0` bytes. Use
  this when streaming sideband output one frame at a time.
  """
  @spec encode_sideband_frame(1 | 2 | 3, binary()) :: binary()
  def encode_sideband_frame(band, chunk)
      when band in [1, 2, 3] and is_binary(chunk) and byte_size(chunk) <= @max_sideband_data do
    encode_raw(<<band, chunk::binary>>)
  end
  # -- Private --
  defp decode_loop(<<>>, acc), do: {:ok, Enum.reverse(acc), <<>>}

@@ -1,0 +1,85 @@
# Copyright 2026 Cole Christensen
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
defmodule ExGitObjectstore.Protocol.SidebandWriter do
  @moduledoc """
  Streaming sideband-1 chunker.
  Wraps a sink callback so callers can push arbitrary-sized binary
  payloads (e.g. pack entries from `ExGitObjectstore.Pack.Writer.generate_stream/2`)
  and have them re-chunked into spec-compliant sideband-1 pkt-line frames
  (≤ 65515 bytes of payload per frame).
  The buffer is held as iodata so each `write/2` is O(1) amortized; we
  only materialize to a binary when we have enough bytes to flush a
  full-size frame. Peak buffer size is bounded by `max_sideband_data + S`
  where `S` is the largest single `write/2` payload.
  """
  alias ExGitObjectstore.Protocol.PktLine
  @max_payload PktLine.max_sideband_data()
  @type sink :: (binary() -> any())
  @type t :: %__MODULE__{
          band: 1 | 2 | 3,
          sink: sink(),
          buf: iodata(),
          buf_bytes: non_neg_integer()
        }
  defstruct band: 1, sink: nil, buf: [], buf_bytes: 0
  @doc """
  Create a new sideband writer that emits frames on the given band
  through `sink_fn`. `sink_fn` receives one full pkt-line frame at a
  time (length prefix + band marker + payload).
  """
  @spec new(1 | 2 | 3, sink()) :: t()
  def new(band, sink_fn) when band in [1, 2, 3] and is_function(sink_fn, 1) do
    %__MODULE__{band: band, sink: sink_fn}
  end
  @doc """
  Append `payload` to the buffer and flush as many complete frames as
  possible. The remainder (< max payload) stays buffered.
  """
  @spec write(t(), iodata()) :: t()
  def write(%__MODULE__{} = state, payload) do
    size = IO.iodata_length(payload)
    drain(%{state | buf: [state.buf, payload], buf_bytes: state.buf_bytes + size})
  end
  @doc """
  Emit any remaining buffered bytes as a final (possibly < max) frame.
  No-op if the buffer is empty.
  """
  @spec finish(t()) :: :ok
  def finish(%__MODULE__{buf_bytes: 0}), do: :ok
  def finish(%__MODULE__{band: band, sink: sink, buf: buf}) do
    sink.(PktLine.encode_sideband_frame(band, IO.iodata_to_binary(buf)))
    :ok
  end
  defp drain(%__MODULE__{buf_bytes: bb} = state) when bb < @max_payload, do: state
  defp drain(%__MODULE__{band: band, sink: sink, buf: buf, buf_bytes: bb} = state) do
    full = IO.iodata_to_binary(buf)
    <<frame::binary-size(@max_payload), rest::binary>> = full
    sink.(PktLine.encode_sideband_frame(band, frame))
    drain(%{state | buf: rest, buf_bytes: bb - @max_payload})
  end
end

@@ -54,6 +54,6 @@
  alias ExGitObjectstore.Object
  alias ExGitObjectstore.Object.{Blob, Commit, Tag, Tree}
  alias ExGitObjectstore.Pack.{Filter, Writer}
  alias ExGitObjectstore.Protocol.PktLine
  alias ExGitObjectstore.Protocol.{PktLine, SidebandWriter}
  @max_tree_depth 64
@@ -99,6 +99,33 @@
    {<<>>, state}
  end
  @doc """
  Feed a v2 command and stream the response bytes through `write_fn`.
  Same behavior as `feed/2`, except the response is delivered as a
  series of `write_fn.(binary)` calls instead of being returned as a
  single binary. Use this for fetch responses: a 161 MB packfile fed
  through `feed/2` peaks at ~10× the pack size in transient BEAM heap
  (pack binary + sideband binary + concatenated response binary); the
  streaming variant bounds peak heap by the largest single object.
  Non-packfile responses (ls-refs, multi-round acks) deliver as a
  single `write_fn` call with the same bytes `feed/2` would return.
  """
  @spec feed(state(), binary(), (binary() -> any())) :: {:ok, state()}
  def feed(%__MODULE__{phase: :command} = state, data, write_fn)
      when is_function(write_fn, 1) do
    buffered = state.buffer <> data
    if has_complete_command?(buffered) do
      process_command_stream(%{state | buffer: <<>>}, buffered, write_fn)
    else
      {:ok, %{state | buffer: buffered}}
    end
  end
  def feed(%__MODULE__{phase: :done} = state, _data, _write_fn), do: {:ok, state}
  defp has_complete_command?(data) do
    # A complete v2 command is terminated by a flush (0000) packet. Delim
    # (0001) is an intra-command section separator between the `command=`
@@ -135,6 +162,33 @@
    end
  end
  defp process_command_stream(state, data, write_fn) do
    case parse_command(data) do
      {:ls_refs, args} ->
        Logger.info("UploadPackV2: processing ls-refs command")
        write_fn.(handle_ls_refs(state.repo, args))
        {:ok, state}
      {:fetch, args} ->
        Logger.info("UploadPackV2: processing fetch command")
        {next_phase, info} = handle_fetch_stream(state.repo, args, write_fn)
        Logger.info(
          "UploadPackV2: fetch streamed #{info.pack_bytes} pack bytes, #{info.objects} objects"
        )
        {:ok, %{state | phase: next_phase}}
      {:error, :no_command} ->
        {:ok, state}
      {:error, err} ->
        Logger.error("UploadPackV2: parse_command failed: #{inspect(err)}")
        write_fn.(PktLine.flush())
        {:ok, %{state | phase: :done}}
    end
  end
  @doc """
  Check if the protocol exchange is complete.
  """
@@ -449,6 +503,151 @@
  defp fetch_mode(nil, _filter, _), do: :filtered
  defp fetch_mode(_shallow, nil, _), do: :shallow
  defp fetch_mode(_, _, _), do: :shallow_filtered
  defp handle_fetch_stream(repo, args, write_fn) do
    case parse_filter_spec(args) do
      {:ok, filter_spec} ->
        handle_fetch_parsed_stream(repo, args, filter_spec, write_fn)
      {:error, err_reply} ->
        write_fn.(err_reply)
        {:done, %{pack_bytes: 0, objects: 0}}
    end
  end
  defp handle_fetch_parsed_stream(repo, args, filter_spec, write_fn) do
    wants = extract_shas(args, "want ")
    haves = extract_shas(args, "have ")
    done? = Enum.any?(args, &(String.trim(&1) == "done"))
    wait_for_done? = Enum.any?(args, &(String.trim(&1) == "wait-for-done"))
    shallow_opts = parse_shallow_opts(args)
    send_packfile? = done? or (shallow_opts != nil and not wait_for_done?)
    Logger.info(
      "UploadPackV2.handle_fetch: #{length(wants)} wants, #{length(haves)} haves, " <>
        "done=#{done?}, wait-for-done=#{wait_for_done?}, " <>
        "shallow=#{shallow_opts != nil}, filter=#{inspect(filter_spec)}, " <>
        "send_packfile=#{send_packfile?}"
    )
    span_meta = %{
      repo_id: repo.id,
      wants: length(wants),
      haves: length(haves),
      mode: fetch_mode(shallow_opts, filter_spec, wait_for_done?)
    }
    :telemetry.span([:ex_git_objectstore, :protocol, :fetch], span_meta, fn ->
      # send_packfile? is derived from {done?, wait_for_done?, shallow_opts}; we
      # recompute it inside do_handle_fetch_stream/8 to keep arity ≤ 8 (credo).
      {next_phase, stats} =
        do_handle_fetch_stream(
          repo,
          wants,
          haves,
          done?,
          wait_for_done?,
          shallow_opts,
          filter_spec,
          write_fn
        )
      {{next_phase, stats}, Map.merge(span_meta, stats)}
    end)
  end
  defp do_handle_fetch_stream(
         repo,
         wants,
         haves,
         done?,
         wait_for_done?,
         shallow_opts,
         filter_spec,
         write_fn
       ) do
    send_packfile? = done? or (shallow_opts != nil and not wait_for_done?)
    cond do
      done? ->
        stats =
          stream_packfile_response(repo, wants, haves, <<>>, shallow_opts, filter_spec, write_fn)
        {:done, stats}
      wait_for_done? ->
        write_fn.(build_acknowledgments(repo, haves, :flush))
        {:command, %{pack_bytes: 0, objects: 0}}
      send_packfile? ->
        ack_section = build_acknowledgments(repo, haves, :ready)
        stats =
          stream_packfile_response(
            repo,
            wants,
            haves,
            ack_section,
            shallow_opts,
            filter_spec,
            write_fn
          )
        {:done, stats}
      true ->
        write_fn.(build_acknowledgments(repo, haves, :flush))
        {:command, %{pack_bytes: 0, objects: 0}}
    end
  end
  defp stream_packfile_response(
         repo,
         wants,
         haves,
         ack_section,
         shallow_opts,
         filter_spec,
         write_fn
       ) do
    case collect_objects_maybe_shallow(repo, wants, haves, shallow_opts, filter_spec) do
      {:ok, %{objects: objects} = walk} ->
        Logger.info("UploadPackV2: collected #{length(objects)} objects, streaming pack")
        shallow_info = build_shallow_info(walk)
        packfile_header = PktLine.encode("packfile")
        # Prefix (ack + shallow + 'packfile' pkt-line) is small — emit as one write.
        write_fn.(IO.iodata_to_binary([ack_section, shallow_info, packfile_header]))
        # Pack body streamed through a sideband-1 chunker. The accumulator
        # carries both the sideband state and a byte counter so we can
        # report pack_bytes for telemetry without holding the full pack.
        sideband = SidebandWriter.new(1, write_fn)
        {_pack_sha, count, {sideband, pack_bytes}} =
          Writer.generate_stream(objects, {sideband, 0}, fn bytes, {sb, n} ->
            {SidebandWriter.write(sb, bytes), n + byte_size(bytes)}
          end)
        :ok = SidebandWriter.finish(sideband)
        # Trailing flush packet ends the response.
        write_fn.(PktLine.flush())
        Logger.info("UploadPackV2: streamed pack #{pack_bytes} bytes, #{count} objects")
        %{pack_bytes: pack_bytes, objects: count}
      {:error, reason} ->
        Logger.error(
          "UploadPackV2: collect_objects failed for #{length(wants)} wants, " <>
            "#{length(haves)} haves: #{inspect(reason)}"
        )
        write_fn.(PktLine.flush())
        %{pack_bytes: 0, objects: 0, error: reason}
    end
  end
  defp build_packfile_response(repo, wants, haves, ack_section, shallow_opts, filter_spec) do
    case collect_objects_maybe_shallow(repo, wants, haves, shallow_opts, filter_spec) do

@@ -105,6 +105,83 @@
    end
  end
  describe "generate_stream" do
    test "byte-equivalent to generate/1 for a single blob" do
      content = "hello world\n"
      blob = Blob.from_content(content)
      sha = Object.hash(blob)
      objects = [{:blob, content, sha}]
      {expected, expected_sha} = Writer.generate(objects)
      {:ok, agent} = Agent.start_link(fn -> [] end)
      {streamed_sha, count} =
        Writer.generate_stream(objects, &Agent.update(agent, fn acc -> [acc, &1] end))
      streamed = agent |> Agent.get(&IO.iodata_to_binary/1)
      Agent.stop(agent)
      assert streamed == expected
      assert streamed_sha == expected_sha
      assert count == 1
    end
    test "byte-equivalent to generate/1 for many mixed objects" do
      blobs =
        for i <- 1..50 do
          content = "content #{i}\n" <> String.duplicate("x", :rand.uniform(2000))
          {:blob, content, Object.hash(Blob.from_content(content))}
        end
      tree = Tree.new([%{mode: "100644", name: "f", sha: String.duplicate("a", 40)}])
      tree_obj = {:tree, Tree.encode_content(tree), Object.hash(tree)}
      objects = [tree_obj | blobs]
      {expected, expected_sha} = Writer.generate(objects)
      {:ok, agent} = Agent.start_link(fn -> [] end)
      {streamed_sha, count} =
        Writer.generate_stream(objects, &Agent.update(agent, fn acc -> [acc, &1] end))
      streamed = agent |> Agent.get(&IO.iodata_to_binary/1)
      Agent.stop(agent)
      assert streamed == expected
      assert streamed_sha == expected_sha
      assert count == length(objects)
    end
    test "/3 threads accumulator through each write" do
      blob = Blob.from_content("threaded\n")
      sha = Object.hash(blob)
      {_sha, count, write_count} =
        Writer.generate_stream([{:blob, "threaded\n", sha}], 0, fn _bytes, n -> n + 1 end)
      # 1 header + 1 entry + 1 checksum = 3 writes
      assert write_count == 3
      assert count == 1
    end
    test "streamed pack parses identically via Reader" do
      blob = Blob.from_content("via reader\n")
      sha = Object.hash(blob)
      objects = [{:blob, "via reader\n", sha}]
      {:ok, agent} = Agent.start_link(fn -> [] end)
      Writer.generate_stream(objects, &Agent.update(agent, fn acc -> [acc, &1] end))
      streamed = agent |> Agent.get(&IO.iodata_to_binary/1)
      Agent.stop(agent)
      assert {:ok, entries} = Reader.parse(streamed)
      assert length(entries) == 1
      assert hd(entries).data == "via reader\n"
    end
  end
  describe "generate_with_index" do
    test "generates both pack and valid index" do
      blob = Blob.from_content("hello\n")

@@ -1,0 +1,107 @@
# Copyright 2026 Cole Christensen
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
defmodule ExGitObjectstore.Protocol.SidebandWriterTest do
  use ExUnit.Case, async: true
  alias ExGitObjectstore.Protocol.{PktLine, SidebandWriter}
  defp collect(fun) do
    {:ok, agent} = Agent.start_link(fn -> [] end)
    sb = SidebandWriter.new(1, fn frame -> Agent.update(agent, fn acc -> [acc, frame] end) end)
    sb = fun.(sb)
    :ok = SidebandWriter.finish(sb)
    frames = agent |> Agent.get(&IO.iodata_to_binary/1)
    Agent.stop(agent)
    frames
  end
  defp decode_sideband_payload(frames) do
    # Concatenate the payloads from every sideband-1 frame in `frames`.
    {:ok, packets, <<>>} = PktLine.decode(frames)
    packets
    |> Enum.map(fn
      {:data, <<1, rest::binary>>} -> rest
    end)
    |> IO.iodata_to_binary()
  end
  defp count_frames(frames) do
    {:ok, packets, <<>>} = PktLine.decode(frames)
    length(packets)
  end
  test "single small write produces one frame with the same payload" do
    frames = collect(fn sb -> SidebandWriter.write(sb, "hello") end)
    assert count_frames(frames) == 1
    assert decode_sideband_payload(frames) == "hello"
  end
  test "empty writer with no payload produces no frames" do
    frames = collect(fn sb -> sb end)
    assert frames == ""
  end
  test "payload larger than max frame is split across multiple frames" do
    max = PktLine.max_sideband_data()
    payload = :crypto.strong_rand_bytes(max * 2 + 100)
    frames = collect(fn sb -> SidebandWriter.write(sb, payload) end)
    {:ok, packets, <<>>} = PktLine.decode(frames)
    # Two full frames + one tail frame.
    assert length(packets) == 3
    assert decode_sideband_payload(frames) == payload
    # Every frame is within spec.
    for {:data, <<_band, chunk::binary>>} <- packets do
      assert byte_size(chunk) <= max
    end
  end
  test "many small writes are coalesced — total bytes preserved, framing within spec" do
    chunks = for _ <- 1..200, do: :crypto.strong_rand_bytes(1000)
    frames =
      collect(fn sb ->
        Enum.reduce(chunks, sb, &SidebandWriter.write(&2, &1))
      end)
    expected = IO.iodata_to_binary(chunks)
    assert decode_sideband_payload(frames) == expected
    max = PktLine.max_sideband_data()
    {:ok, packets, <<>>} = PktLine.decode(frames)
    for {:data, <<_band, chunk::binary>>} <- packets do
      assert byte_size(chunk) <= max
    end
  end
  test "writes spanning exactly a frame boundary do not buffer partial bytes" do
    max = PktLine.max_sideband_data()
    chunks = [String.duplicate("a", max), "remainder"]
    frames =
      collect(fn sb ->
        Enum.reduce(chunks, sb, &SidebandWriter.write(&2, &1))
      end)
    {:ok, packets, <<>>} = PktLine.decode(frames)
    # First write fills a frame exactly (drained), second is a short final frame.
    assert length(packets) == 2
    assert decode_sideband_payload(frames) == IO.iodata_to_binary(chunks)
  end
end

@@ -739,6 +739,168 @@
    end
  end
  describe "streaming feed/3" do
    defp collect_stream(state, data) do
      {:ok, agent} = Agent.start_link(fn -> [] end)
      {:ok, new_state} =
        UploadPackV2.feed(state, data, fn chunk ->
          Agent.update(agent, fn acc -> [acc, chunk] end)
        end)
      streamed = Agent.get(agent, &IO.iodata_to_binary/1)
      Agent.stop(agent)
      {streamed, new_state}
    end
    test "ls-refs response is byte-identical to feed/2" do
      repo = RepoHelper.memory_repo()
      ExGitObjectstore.init(repo)
      {commit_sha, _, _} = create_commit(repo, "x\n", "init\n")
      :ok = Ref.put(repo, "refs/heads/main", commit_sha, nil)
      client_data =
        PktLine.encode("command=ls-refs") <> PktLine.delim() <> PktLine.flush()
      {_advert, s1} = UploadPackV2.init(repo)
      {_advert, s2} = UploadPackV2.init(repo)
      {expected, new_s1} = UploadPackV2.feed(s1, client_data)
      {streamed, new_s2} = collect_stream(s2, client_data)
      assert streamed == expected
      assert new_s1.phase == new_s2.phase
    end
    test "fetch response (clone) is byte-identical to feed/2" do
      repo = RepoHelper.memory_repo()
      ExGitObjectstore.init(repo)
      {shas, _} = create_chain(repo, 5)
      tip = List.last(shas)
      client_data =
        PktLine.encode("command=fetch") <>
          PktLine.delim() <>
          PktLine.encode("want #{tip}") <>
          PktLine.encode("done") <>
          PktLine.flush()
      {_advert, s1} = UploadPackV2.init(repo)
      {_advert, s2} = UploadPackV2.init(repo)
      {expected, new_s1} = UploadPackV2.feed(s1, client_data)
      {streamed, new_s2} = collect_stream(s2, client_data)
      assert byte_size(streamed) == byte_size(expected)
      assert streamed == expected
      assert new_s2.phase == :done
      assert new_s1.phase == :done
    end
    test "multi-round fetch ack section is byte-identical to feed/2" do
      repo = RepoHelper.memory_repo()
      ExGitObjectstore.init(repo)
      {shas, _} = create_chain(repo, 3)
      tip = List.last(shas)
      [first | _] = shas
      # haves without `done` → server should emit an acks-only section.
      client_data =
        PktLine.encode("command=fetch") <>
          PktLine.delim() <>
          PktLine.encode("want #{tip}") <>
          PktLine.encode("have #{first}") <>
          PktLine.flush()
      {_advert, s1} = UploadPackV2.init(repo)
      {_advert, s2} = UploadPackV2.init(repo)
      {expected, new_s1} = UploadPackV2.feed(s1, client_data)
      {streamed, new_s2} = collect_stream(s2, client_data)
      assert streamed == expected
      assert new_s1.phase == :command
      assert new_s2.phase == :command
    end
    test "extracted pack from streamed response parses correctly" do
      repo = RepoHelper.memory_repo()
      ExGitObjectstore.init(repo)
      {commit_sha, _, _} = create_commit(repo, "hello stream\n", "init\n")
      :ok = Ref.put(repo, "refs/heads/main", commit_sha, nil)
      client_data =
        PktLine.encode("command=fetch") <>
          PktLine.delim() <>
          PktLine.encode("want #{commit_sha}") <>
          PktLine.encode("done") <>
          PktLine.flush()
      {_advert, state} = UploadPackV2.init(repo)
      {streamed, _new_state} = collect_stream(state, client_data)
      pack = extract_sideband_pack(streamed)
      assert pack != nil
      assert {:ok, entries} = Reader.parse(pack)
      assert length(entries) >= 3
      assert Enum.any?(entries, &(&1.type == :commit))
    end
    test "streamed fetch emits multiple chunks, not one giant binary" do
      repo = RepoHelper.memory_repo()
      ExGitObjectstore.init(repo)
      # Build a repo big enough to push pack body over a single sideband frame.
      # max_sideband_data = 65515; one 70 KiB blob gives us multiple frames.
      blob = Blob.from_content(:crypto.strong_rand_bytes(70 * 1024))
      {:ok, blob_sha} = Object.write(repo, blob)
      tree = Tree.new([%{mode: "100644", name: "big", sha: blob_sha}])
      {:ok, tree_sha} = Object.write(repo, tree)
      commit = %Commit{
        tree: tree_sha,
        parents: [],
        author: "T <t@t> 1 +0000",
        committer: "T <t@t> 1 +0000",
        message: "big\n"
      }
      {:ok, commit_sha} = Object.write(repo, commit)
      :ok = Ref.put(repo, "refs/heads/main", commit_sha, nil)
      client_data =
        PktLine.encode("command=fetch") <>
          PktLine.delim() <>
          PktLine.encode("want #{commit_sha}") <>
          PktLine.encode("done") <>
          PktLine.flush()
      {_advert, state} = UploadPackV2.init(repo)
      {:ok, agent} = Agent.start_link(fn -> [] end)
      {:ok, _new_state} =
        UploadPackV2.feed(state, client_data, fn chunk ->
          Agent.update(agent, fn acc -> [{byte_size(chunk), chunk} | acc] end)
        end)
      chunks = Agent.get(agent, & &1) |> Enum.reverse()
      Agent.stop(agent)
      # Should be at least: prefix + 1 sideband frame + trailing flush = 3 writes.
      assert length(chunks) >= 3
      # No single chunk should exceed the pkt-line max (65520 bytes).
      for {size, _} <- chunks do
        assert size <= 65_520
      end
      # Recomposed bytes still produce a valid pack.
      recomposed = chunks |> Enum.map(fn {_, c} -> c end) |> IO.iodata_to_binary()
      pack = extract_sideband_pack(recomposed)
      assert {:ok, _entries} = Reader.parse(pack)
    end
  end
  # --- Helpers ---
  # Extract pack data from v2 fetch response sideband encoding.