"""TDD tests for MPack integration in transport.py.

These tests verify that HttpTransport.push_stream and fetch_stream use the
MPackStreamWriter/Reader API (muse.core.mpack) rather than the old inline
frame construction, and that LocalFileTransport has a push_stream method.

Red phase: all tests below are written against the *desired* behaviour.
Tests marked with their expected failure reason:
- content-type tests: current code uses "application/x-muse-packstream"
- H-frame "v" tests: current H frame omits the version field
- O-frame "sz" tests: current O frame omits the uncompressed size field
- compression tests: current code hardcodes zlib, not choose_compression()
- zstd decompression: fetch_stream only handles zlib today
- LocalFileTransport.push_stream: method does not exist yet
"""

from __future__ import annotations

import hashlib
import json
import pathlib
import unittest.mock
from io import BytesIO

import msgpack
import pytest

from muse.core._types import long_id
from muse.core.mpack import (
    MPACK_CONTENT_TYPE,
    MPACK_VERSION,
    MPackStreamReader,
    MPackStreamWriter,
)
from muse.core.transport import (
    HttpTransport,
    LocalFileTransport,
    PushResult,
    SigningIdentity,
    TransportError,
)


# ---------------------------------------------------------------------------
# Shared helpers
# ---------------------------------------------------------------------------


def _make_signing() -> SigningIdentity:
    from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
    return SigningIdentity(handle="testuser", private_key=Ed25519PrivateKey.generate())


def _mock_stream_response(
    frames_bytes: bytes,
    status: int = 200,
    content_type: str = MPACK_CONTENT_TYPE,
) -> unittest.mock.MagicMock:
    """Streaming response that returns frames_bytes then b""."""
    resp = unittest.mock.MagicMock()
    resp.read.side_effect = [frames_bytes, b""]
    resp.headers = {"Content-Type": content_type}
    resp.__enter__ = lambda s: s
    resp.__exit__ = unittest.mock.MagicMock(return_value=False)
    return resp


class _MockPushConn:
    """Minimal http.client.HTTPConnection stand-in for push_stream tests."""

    def __init__(self, response_body: bytes) -> None:
        self.headers: dict[str, str] = {}
        self.sends: list[bytes] = []
        self._response_body = response_body
        self.closed = False

    def putrequest(self, method: str, path: str, **kw: object) -> None:
        pass

    def putheader(self, key: str, value: str) -> None:
        self.headers[key.lower()] = value

    def endheaders(self, body: bytes | None = None) -> None:
        pass

    def send(self, data: bytes) -> None:
        self.sends.append(bytes(data))

    def getresponse(self) -> object:
        class _Resp:
            def __init__(self, body: bytes) -> None:
                self.status = 200
                self._buf = body
                self._pos = 0

            def read(self, n: int = -1) -> bytes:
                if n == -1:
                    out = self._buf[self._pos:]
                    self._pos = len(self._buf)
                    return out
                out = self._buf[self._pos : self._pos + n]
                self._pos += n
                return out

            def getheader(self, name: str, default: str | None = None) -> str | None:
                return default
        return _Resp(self._response_body)

    def close(self) -> None:
        self.closed = True


def _decode_push_frames(mock_conn: _MockPushConn) -> list[dict]:
    """Reassemble frame bytes from chunked sends and decode MPack frames."""
    import msgpack as _msgpack
    frames = []
    for raw in mock_conn.sends:
        if raw == b"0\r\n\r\n":
            continue
        try:
            # HTTP chunk: {hex_size}\r\n{data}\r\n
            nl = raw.index(b"\r\n")
            payload = raw[nl + 2 : -2]
            obj = _msgpack.unpackb(payload, raw=False)
            if isinstance(obj, dict):
                frames.append(obj)
        except Exception:  # noqa: BLE001
            pass
    return frames


def _capture_push_request(
    transport: HttpTransport,
    *,
    objects: list | None = None,
    commits: list | None = None,
    snapshots: list | None = None,
    branch: str = "main",
    force: bool = False,
    have: list | None = None,
    local_head: str | None = None,
    server_result: bytes | None = None,
) -> dict:
    """Call push_stream via chunked _open_chunked_connection seam and capture state.

    Returns a dict with keys:
        content_type — Content-Type header sent on connection
        accept       — Accept header sent on connection
        frames       — list of decoded MPack frame dicts (from chunked sends)
    """
    writer = MPackStreamWriter()
    result_frame = writer.write_result(ok=True, msg="ok", head=long_id("a" * 64))
    response_body = server_result or result_frame

    mock_conn = _MockPushConn(response_body)

    def factory(host, port, *, use_ssl, timeout, context=None):
        return mock_conn

    with unittest.mock.patch(
        "muse.core.transport._open_chunked_connection", side_effect=factory
    ):
        try:
            transport.push_stream(
                url="http://localhost:10003",
                signing=None,
                objects=objects or [],
                commits=commits or [],
                snapshots=snapshots or [],
                branch=branch,
                force=force,
                have=have or [],
                local_head=local_head,
            )
        except TransportError:
            pass

    captured: dict = {
        "content_type": mock_conn.headers.get("content-type", ""),
        "accept": mock_conn.headers.get("accept", ""),
        "frames": _decode_push_frames(mock_conn),
    }
    return captured


def _run_push_via_chunked(
    transport: HttpTransport,
    *,
    objects: list | None = None,
    commits: list | None = None,
    snapshots: list | None = None,
    branch: str = "main",
    signing: SigningIdentity | None = None,
    response_body: bytes | None = None,
    expect_error: bool = False,
) -> tuple[_MockPushConn, dict]:
    """Invoke push_stream via the chunked connection seam. Returns (conn, result)."""
    writer = MPackStreamWriter()
    default_resp = writer.write_result(ok=True, msg="ok", head=long_id("a" * 64))
    body = response_body or default_resp
    conn = _MockPushConn(body)

    def factory(host, port, *, use_ssl, timeout, context=None):
        return conn

    result: dict = {}
    with unittest.mock.patch(
        "muse.core.transport._open_chunked_connection", side_effect=factory
    ):
        try:
            result = transport.push_stream(
                url="http://localhost:10003",
                signing=signing,
                objects=objects or [],
                commits=commits or [],
                snapshots=snapshots or [],
                branch=branch,
                force=False,
                have=[],
            )
        except TransportError:
            if not expect_error:
                raise
    return conn, result


def _capture_fetch_request(
    transport: HttpTransport,
    *,
    want: list | None = None,
    have: list | None = None,
    server_frames: bytes | None = None,
) -> dict:
    """Call fetch_stream and capture outgoing request headers."""
    writer = MPackStreamWriter()
    end_frame = writer.write_end(n_objects=0, n_commits=0)
    commit_frame = writer.write_commit_pack(commits=[], snapshots=[])
    h_frame = writer.write_header(
        op="fetch", branch="main", n_objects=0, n_commits=0,
        branch_heads={}, repo_id="r", domain="code", default_branch="main",
    )
    response_body = server_frames or (h_frame + commit_frame + end_frame)

    captured: dict = {}

    def fake_open(req, timeout):
        captured["content_type"] = req.get_header("Content-type")
        captured["accept"] = req.get_header("Accept")
        captured["body"] = req.data
        return _mock_stream_response(response_body)

    with unittest.mock.patch(
        "muse.core.transport._open_url", side_effect=fake_open
    ):
        try:
            transport.fetch_stream(
                url="http://localhost:10003",
                signing=None,
                want=want or [],
                have=have or [],
            )
        except (TransportError, Exception):
            pass

    return captured


def _make_repo(tmp_path: pathlib.Path) -> pathlib.Path:
    repo = tmp_path / "repo"
    muse = repo / ".muse"
    for sub in ("objects", "commits", "snapshots", "refs/heads"):
        (muse / sub).mkdir(parents=True)
    (muse / "HEAD").write_text("ref: refs/heads/main")
    (muse / "repo.json").write_text(
        json.dumps({"repo_id": "test-repo", "domain": "code", "default_branch": "main"})
    )
    return repo


# ---------------------------------------------------------------------------
# HttpTransport.push_stream — content-type
# ---------------------------------------------------------------------------


class TestPushStreamContentType:
    def test_push_stream_uses_mpack_content_type(self) -> None:
        """push_stream must set Content-Type: application/x-muse-mpack."""
        t = HttpTransport()
        cap = _capture_push_request(t)
        assert cap.get("content_type", "").lower() == MPACK_CONTENT_TYPE.lower(), (
            f"Expected Content-Type {MPACK_CONTENT_TYPE!r}, got {cap.get('content_type')!r}"
        )

    def test_push_stream_accept_uses_mpack_content_type(self) -> None:
        """push_stream must set Accept: application/x-muse-mpack."""
        t = HttpTransport()
        cap = _capture_push_request(t)
        assert cap.get("accept", "").lower() == MPACK_CONTENT_TYPE.lower(), (
            f"Expected Accept {MPACK_CONTENT_TYPE!r}, got {cap.get('accept')!r}"
        )


# ---------------------------------------------------------------------------
# HttpTransport.push_stream — H frame
# ---------------------------------------------------------------------------


class TestPushStreamHFrame:
    def test_h_frame_is_first(self) -> None:
        """First frame must be H."""
        t = HttpTransport()
        cap = _capture_push_request(t)
        frames = cap["frames"]
        assert frames, "No frames decoded from push body"
        assert frames[0].get("t") == "H", f"Expected H frame first, got {frames[0].get('t')!r}"

    def test_h_frame_has_protocol_version(self) -> None:
        """H frame must carry v=MPACK_VERSION."""
        t = HttpTransport()
        cap = _capture_push_request(t)
        h = next((f for f in cap["frames"] if f.get("t") == "H"), None)
        assert h is not None, "No H frame in push body"
        assert h.get("v") == MPACK_VERSION, (
            f"H frame missing v={MPACK_VERSION!r}, got v={h.get('v')!r}"
        )

    def test_h_frame_has_op_push(self) -> None:
        """H frame op must be 'push'."""
        t = HttpTransport()
        cap = _capture_push_request(t)
        h = next((f for f in cap["frames"] if f.get("t") == "H"), None)
        assert h is not None
        assert h.get("op") == "push", f"H frame op={h.get('op')!r}, expected 'push'"

    def test_h_frame_branch_matches(self) -> None:
        """H frame branch must match the requested branch."""
        t = HttpTransport()
        cap = _capture_push_request(t, branch="dev")
        h = next((f for f in cap["frames"] if f.get("t") == "H"), None)
        assert h is not None
        assert h.get("branch") == "dev"

    def test_h_frame_n_objects_advisory_count(self) -> None:
        """H frame n_objects must equal the number of objects passed."""
        t = HttpTransport()
        content = b"hello mpack"
        oid = long_id(hashlib.sha256(content).hexdigest())
        objects = [{"object_id": oid, "content": content, "path": "a.txt", "encoding": "raw"}]
        cap = _capture_push_request(t, objects=objects)
        h = next((f for f in cap["frames"] if f.get("t") == "H"), None)
        assert h is not None
        assert h.get("n_objects") == 1


# ---------------------------------------------------------------------------
# HttpTransport.push_stream — O frames
# ---------------------------------------------------------------------------


class TestPushStreamOFrames:
    def test_o_frame_has_sz_field(self) -> None:
        """O frame must carry sz = uncompressed byte length."""
        t = HttpTransport()
        content = b"object content for sz test"
        oid = long_id(hashlib.sha256(content).hexdigest())
        objects = [{"object_id": oid, "content": content, "path": "f.txt", "encoding": "raw"}]
        cap = _capture_push_request(t, objects=objects)
        o_frames = [f for f in cap["frames"] if f.get("t") == "O"]
        assert o_frames, "No O frames in push body"
        for o in o_frames:
            assert "sz" in o, f"O frame missing 'sz' field: {list(o.keys())}"
            assert o["sz"] == len(content), (
                f"O frame sz={o['sz']}, expected {len(content)}"
            )

    def test_o_frame_id_matches_object_id(self) -> None:
        """O frame 'id' must match the object_id."""
        t = HttpTransport()
        content = b"id match test"
        oid = long_id(hashlib.sha256(content).hexdigest())
        objects = [{"object_id": oid, "content": content, "path": "g.txt", "encoding": "raw"}]
        cap = _capture_push_request(t, objects=objects)
        o_frames = [f for f in cap["frames"] if f.get("t") == "O"]
        assert o_frames
        assert o_frames[0].get("id") == oid

    def test_o_frame_content_decompresses_to_original(self) -> None:
        """O frame content must decompress back to the original bytes."""
        t = HttpTransport()
        content = b"round-trip compression test " * 20
        oid = long_id(hashlib.sha256(content).hexdigest())
        objects = [{"object_id": oid, "content": content, "path": "h.txt", "encoding": "raw"}]
        cap = _capture_push_request(t, objects=objects)
        o_frames = [f for f in cap["frames"] if f.get("t") == "O"]
        assert o_frames
        reader = MPackStreamReader()
        recovered = reader.decompress_object(o_frames[0])
        assert recovered == content, "Decompressed O frame content does not match original"

    def test_o_frame_uses_best_compression(self) -> None:
        """O frame enc must be whatever choose_compression() returns (zstd or zlib)."""
        from muse.core.compression import choose_compression
        t = HttpTransport()
        content = b"compression selection test " * 10
        oid = long_id(hashlib.sha256(content).hexdigest())
        objects = [{"object_id": oid, "content": content, "path": "i.txt", "encoding": "raw"}]
        cap = _capture_push_request(t, objects=objects)
        o_frames = [f for f in cap["frames"] if f.get("t") == "O"]
        assert o_frames
        expected_enc = choose_compression()
        assert o_frames[0].get("enc") == expected_enc, (
            f"O frame enc={o_frames[0].get('enc')!r}, expected {expected_enc!r} "
            f"from choose_compression()"
        )

    def test_multiple_objects_all_have_sz(self) -> None:
        """All O frames in a multi-object push must have sz."""
        t = HttpTransport()
        objects = []
        for i in range(5):
            content = f"object-{i}".encode() * 10
            oid = long_id(hashlib.sha256(content).hexdigest())
            objects.append({"object_id": oid, "content": content, "path": f"{i}.txt", "encoding": "raw"})
        cap = _capture_push_request(t, objects=objects)
        o_frames = [f for f in cap["frames"] if f.get("t") == "O"]
        assert len(o_frames) == 5
        for o in o_frames:
            assert "sz" in o, f"O frame missing sz: {list(o.keys())}"


# ---------------------------------------------------------------------------
# HttpTransport.push_stream — frame sequence
# ---------------------------------------------------------------------------


class TestPushStreamFrameSequence:
    def test_frame_sequence_h_c_e(self) -> None:
        """Frame sequence for empty push must be H → C → E."""
        t = HttpTransport()
        cap = _capture_push_request(t)
        tags = [f.get("t") for f in cap["frames"]]
        assert "H" in tags, f"Missing H frame. Tags: {tags}"
        assert "C" in tags, f"Missing C frame. Tags: {tags}"
        assert "E" in tags, f"Missing E frame. Tags: {tags}"
        assert tags.index("H") < tags.index("C") < tags.index("E"), (
            f"Frame order wrong: {tags}"
        )

    def test_frame_sequence_h_o_c_e_with_objects(self) -> None:
        """Frame sequence with objects must be H → O... → C → E."""
        t = HttpTransport()
        content = b"seq test"
        oid = long_id(hashlib.sha256(content).hexdigest())
        objects = [{"object_id": oid, "content": content, "path": "s.txt", "encoding": "raw"}]
        cap = _capture_push_request(t, objects=objects)
        tags = [f.get("t") for f in cap["frames"]]
        assert tags[0] == "H", f"First frame must be H, got {tags[0]!r}"
        assert tags[-1] == "E", f"Last frame must be E, got {tags[-1]!r}"
        assert "O" in tags
        assert "C" in tags
        h_idx = tags.index("H")
        o_idx = tags.index("O")
        c_idx = tags.index("C")
        e_idx = tags.index("E")
        assert h_idx < o_idx < c_idx < e_idx, f"Frame order wrong: {tags}"

    def test_e_frame_n_objects_matches_actual(self) -> None:
        """E frame n_objects must equal the actual number of O frames sent."""
        t = HttpTransport()
        objects = []
        for i in range(3):
            content = f"obj-{i}".encode()
            oid = long_id(hashlib.sha256(content).hexdigest())
            objects.append({"object_id": oid, "content": content, "path": f"{i}.txt", "encoding": "raw"})
        cap = _capture_push_request(t, objects=objects)
        e = next((f for f in cap["frames"] if f.get("t") == "E"), None)
        assert e is not None, "No E frame found"
        assert e.get("n_objects") == 3, f"E frame n_objects={e.get('n_objects')}, expected 3"


# ---------------------------------------------------------------------------
# HttpTransport.push_stream — response parsing
# ---------------------------------------------------------------------------


class TestPushStreamResponseParsing:
    def test_progress_frame_does_not_raise(self) -> None:
        """P frames in server response must not raise."""
        t = HttpTransport()
        writer = MPackStreamWriter()
        response = (
            writer.write_progress(msg="resolving objects", pct=50.0)
            + writer.write_result(ok=True, msg="ok", head=long_id("a" * 64))
        )
        # Should not raise
        _capture_push_request(t, server_result=response)

    def test_error_frame_raises_transport_error(self) -> None:
        """X frame in server response must raise TransportError."""
        t = HttpTransport()
        writer = MPackStreamWriter()
        response = writer.write_error(msg="conflict", code=409)
        _, result_dict = _run_push_via_chunked(t, response_body=response, expect_error=True)
        # error path — TransportError is caught; result_dict is empty
        assert result_dict == {}

    def test_error_frame_status_code(self) -> None:
        """X frame code=409 must become TransportError with status_code=409."""
        t = HttpTransport()
        writer = MPackStreamWriter()
        response = writer.write_error(msg="conflict", code=409)

        def factory(host, port, *, use_ssl, timeout, context=None):
            return _MockPushConn(response)

        with unittest.mock.patch(
            "muse.core.transport._open_chunked_connection", side_effect=factory
        ):
            with pytest.raises(TransportError) as exc_info:
                t.push_stream(
                    url="http://localhost:10003",
                    signing=None,
                    objects=[],
                    commits=[],
                    snapshots=[],
                    branch="main",
                    force=False,
                    have=[],
                )
        assert exc_info.value.status_code == 409

    def test_result_frame_ok_returns_push_result(self) -> None:
        """R frame with ok=True returns PushResult(ok=True)."""
        t = HttpTransport()
        writer = MPackStreamWriter()
        tip = long_id("b" * 64)
        response = writer.write_result(ok=True, msg="pushed", heads={"main": tip}, head=tip)
        _, result = _run_push_via_chunked(t, response_body=response)
        assert isinstance(result, dict)
        assert result["ok"] is True


# ---------------------------------------------------------------------------
# HttpTransport.fetch_stream — content-type
# ---------------------------------------------------------------------------


class TestFetchStreamContentType:
    def test_fetch_stream_uses_mpack_content_type(self) -> None:
        """fetch_stream must set Content-Type: application/x-muse-mpack."""
        t = HttpTransport()
        cap = _capture_fetch_request(t)
        assert cap.get("content_type", "").lower() == MPACK_CONTENT_TYPE.lower(), (
            f"Expected {MPACK_CONTENT_TYPE!r}, got {cap.get('content_type')!r}"
        )

    def test_fetch_stream_accept_uses_mpack_content_type(self) -> None:
        """fetch_stream must set Accept: application/x-muse-mpack."""
        t = HttpTransport()
        cap = _capture_fetch_request(t)
        assert cap.get("accept", "").lower() == MPACK_CONTENT_TYPE.lower(), (
            f"Expected {MPACK_CONTENT_TYPE!r}, got {cap.get('accept')!r}"
        )


# ---------------------------------------------------------------------------
# HttpTransport.fetch_stream — O frame decompression
# ---------------------------------------------------------------------------


class TestFetchStreamDecompression:
    def _make_fetch_response_with_object(self, enc: str, content: bytes, oid: str) -> bytes:
        """Build a fetch response stream containing one O frame with given enc."""
        from muse.core.compression import compress_zlib, compress_zstd
        writer = MPackStreamWriter()
        if enc == "zstd":
            wire = compress_zstd(content)
        elif enc == "zlib":
            wire = compress_zlib(content)
        else:
            wire = content

        h = writer.write_header(
            op="fetch", branch="main", n_objects=1, n_commits=0,
            repo_id="r", domain="code", default_branch="main", branch_heads={},
        )
        o = writer.write_object(
            object_id=oid, content=wire, enc=enc, sz=len(content), path="a.txt",
        )
        c = writer.write_commit_pack(commits=[], snapshots=[])
        e = writer.write_end(n_objects=1, n_commits=0)
        return h + o + c + e

    def test_fetch_stream_decompresses_zlib_objects(self) -> None:
        """fetch_stream must decompress zlib-encoded O frames correctly."""
        content = b"zlib content for fetch " * 5
        oid = long_id(hashlib.sha256(content).hexdigest())
        server_frames = self._make_fetch_response_with_object("zlib", content, oid)

        received: list = []
        t = HttpTransport()
        with unittest.mock.patch(
            "muse.core.transport._open_url",
            return_value=_mock_stream_response(server_frames),
        ):
            t.fetch_stream(
                url="http://localhost:10003",
                signing=None,
                want=[oid],
                have=[],
                on_object=received.append,
            )
        assert len(received) == 1
        assert received[0]["content"] == content

    def test_fetch_stream_decompresses_zstd_objects(self) -> None:
        """fetch_stream must decompress zstd-encoded O frames (via MPackStreamReader)."""
        from muse.core.compression import ZSTD_AVAILABLE
        if not ZSTD_AVAILABLE:
            pytest.skip("zstd not available on this machine")

        content = b"zstd content for fetch " * 5
        oid = long_id(hashlib.sha256(content).hexdigest())
        server_frames = self._make_fetch_response_with_object("zstd", content, oid)

        received: list = []
        t = HttpTransport()
        with unittest.mock.patch(
            "muse.core.transport._open_url",
            return_value=_mock_stream_response(server_frames),
        ):
            t.fetch_stream(
                url="http://localhost:10003",
                signing=None,
                want=[oid],
                have=[],
                on_object=received.append,
            )
        assert len(received) == 1, f"Expected 1 object, got {len(received)}"
        assert received[0]["content"] == content, "Decompressed content mismatch"

    def test_fetch_stream_objects_received_count(self) -> None:
        """fetch_stream result.objects_received must equal actual O frames."""
        content = b"count test"
        oid = long_id(hashlib.sha256(content).hexdigest())
        server_frames = self._make_fetch_response_with_object("zlib", content, oid)

        t = HttpTransport()
        with unittest.mock.patch(
            "muse.core.transport._open_url",
            return_value=_mock_stream_response(server_frames),
        ):
            result = t.fetch_stream(
                url="http://localhost:10003",
                signing=None,
                want=[oid],
                have=[],
            )
        assert result["objects_received"] == 1


# ---------------------------------------------------------------------------
# LocalFileTransport.push_stream
# ---------------------------------------------------------------------------


class TestLocalFileTransportPushStream:
    def test_push_stream_method_exists(self) -> None:
        """LocalFileTransport must have a push_stream method."""
        t = LocalFileTransport()
        assert hasattr(t, "push_stream"), (
            "LocalFileTransport is missing push_stream — method not implemented"
        )
        assert callable(t.push_stream)

    def test_push_stream_writes_commit_to_remote(self, tmp_path: pathlib.Path) -> None:
        """push_stream must write commits to the remote store."""
        import datetime
        from muse.core.snapshot import compute_commit_id, compute_snapshot_id
        from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot

        src = _make_repo(tmp_path / "src")
        dst = _make_repo(tmp_path / "dst")

        snap_id = compute_snapshot_id({})
        write_snapshot(src, SnapshotRecord(
            snapshot_id=snap_id,
            manifest={},
            created_at=datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc),
        ))
        committed_at = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
        commit_id = compute_commit_id([], snap_id, "test push_stream", committed_at.isoformat())
        write_commit(src, CommitRecord(
            commit_id=commit_id,
            repo_id="test-repo",
            branch="main",
            snapshot_id=snap_id,
            message="test push_stream",
            committed_at=committed_at,
            parent_commit_id=None,
        ))

        # Build commits/snapshots as dicts to pass to push_stream
        commits = [{"commit_id": commit_id, "repo_id": "test-repo", "branch": "main",
                    "snapshot_id": snap_id, "message": "test push_stream",
                    "committed_at": "2026-01-01T00:00:00+00:00", "parent_commit_id": None,
                    "parent2_commit_id": None, "agent_id": "", "model_id": "", "tags": []}]
        snapshots = [{"snapshot_id": snap_id, "manifest": {},
                      "created_at": "2026-01-01T00:00:00+00:00"}]

        t = LocalFileTransport()
        result = t.push_stream(
            url=dst.as_uri(),
            signing=None,
            objects=[],
            commits=commits,
            snapshots=snapshots,
            branch="main",
            force=False,
            have=[],
            local_head=commit_id,
        )
        assert isinstance(result, dict)
        assert result["ok"], f"push_stream returned ok=False: {result.get('message')}"

        # Commit must exist in dst store
        from muse.core.store import _commit_path
        dst_commit_path = _commit_path(dst, commit_id)
        assert dst_commit_path.exists(), (
            f"Commit not found in dst store at {dst_commit_path}"
        )

    def test_push_stream_writes_objects_to_remote(self, tmp_path: pathlib.Path) -> None:
        """push_stream must write objects to the remote object store."""
        import datetime
        from muse.core.snapshot import compute_commit_id, compute_snapshot_id
        from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot

        src = _make_repo(tmp_path / "src")
        dst = _make_repo(tmp_path / "dst")

        content = b"binary object for push_stream"
        oid = long_id(hashlib.sha256(content).hexdigest())

        snap_id = compute_snapshot_id({"file.bin": oid})
        write_snapshot(src, SnapshotRecord(
            snapshot_id=snap_id,
            manifest={"file.bin": oid},
            created_at=datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc),
        ))
        committed_at = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
        commit_id = compute_commit_id([], snap_id, "obj test", committed_at.isoformat())
        write_commit(src, CommitRecord(
            commit_id=commit_id,
            repo_id="test-repo",
            branch="main",
            snapshot_id=snap_id,
            message="obj test",
            committed_at=committed_at,
            parent_commit_id=None,
        ))

        commits = [{"commit_id": commit_id, "repo_id": "test-repo", "branch": "main",
                    "snapshot_id": snap_id, "message": "obj test",
                    "committed_at": "2026-01-01T00:00:00+00:00", "parent_commit_id": None,
                    "parent2_commit_id": None, "agent_id": "", "model_id": "", "tags": []}]
        snapshots = [{"snapshot_id": snap_id, "manifest": {"file.bin": oid},
                      "created_at": "2026-01-01T00:00:00+00:00"}]
        objects = [{"object_id": oid, "content": content, "path": "file.bin", "encoding": "raw"}]

        t = LocalFileTransport()
        result = t.push_stream(
            url=dst.as_uri(),
            signing=None,
            objects=objects,
            commits=commits,
            snapshots=snapshots,
            branch="main",
            force=False,
            have=[],
            local_head=commit_id,
        )
        assert result["ok"], f"push_stream returned ok=False: {result.message}"

        # Object must exist in dst object store
        from muse.core.object_store import object_path
        dst_obj = object_path(dst, oid)
        assert dst_obj.exists(), f"Object not found in dst at {dst_obj}"

    def test_push_stream_updates_branch_ref(self, tmp_path: pathlib.Path) -> None:
        """push_stream must update the remote branch ref to the new tip."""
        import datetime
        from muse.core.snapshot import compute_commit_id, compute_snapshot_id
        from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot

        src = _make_repo(tmp_path / "src")
        dst = _make_repo(tmp_path / "dst")

        snap_id = compute_snapshot_id({})
        write_snapshot(src, SnapshotRecord(
            snapshot_id=snap_id,
            manifest={},
            created_at=datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc),
        ))
        committed_at = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
        commit_id = compute_commit_id([], snap_id, "branch ref test", committed_at.isoformat())
        write_commit(src, CommitRecord(
            commit_id=commit_id,
            repo_id="test-repo",
            branch="main",
            snapshot_id=snap_id,
            message="branch ref test",
            committed_at=committed_at,
            parent_commit_id=None,
        ))

        commits = [{"commit_id": commit_id, "repo_id": "test-repo", "branch": "main",
                    "snapshot_id": snap_id, "message": "branch ref test",
                    "committed_at": "2026-01-01T00:00:00+00:00", "parent_commit_id": None,
                    "parent2_commit_id": None, "agent_id": "", "model_id": "", "tags": []}]
        snapshots = [{"snapshot_id": snap_id, "manifest": {},
                      "created_at": "2026-01-01T00:00:00+00:00"}]

        t = LocalFileTransport()
        result = t.push_stream(
            url=dst.as_uri(),
            signing=None,
            objects=[],
            commits=commits,
            snapshots=snapshots,
            branch="main",
            force=False,
            have=[],
            local_head=commit_id,
        )
        assert result["ok"]

        # Branch ref must point to the commit
        ref_path = dst / ".muse" / "refs" / "heads" / "main"
        assert ref_path.exists(), "Branch ref not created"
        assert ref_path.read_text().strip() == commit_id

    def test_push_stream_returns_push_result(self, tmp_path: pathlib.Path) -> None:
        """push_stream must return a PushResult (dict with ok, message, branch_heads)."""
        dst = _make_repo(tmp_path / "dst")
        t = LocalFileTransport()
        result = t.push_stream(
            url=dst.as_uri(),
            signing=None,
            objects=[],
            commits=[],
            snapshots=[],
            branch="main",
            force=False,
            have=[],
        )
        assert isinstance(result, dict)
        assert "ok" in result
        assert "message" in result or "branch_heads" in result