diff --git a/syft/sync/messages/raw_message.py b/syft/sync/messages/raw_message.py
index ad5077f589a..6ce9a329196 100644
--- a/syft/sync/messages/raw_message.py
+++ b/syft/sync/messages/raw_message.py
@@ -1,13 +1,5 @@
-from pydantic import BaseModel
+"""``RawMessage`` lives with the other transport types; this path keeps working."""
 
+from syft.sync.transport.types import RawMessage
 
-class RawMessage(BaseModel):
-    """Generic container for downloaded bytes + metadata.
-
-    Transport returns this — no deserialization, no encryption awareness.
-    """
-
-    data: bytes
-    sender_email: str | None = None
-    platform_id: str | None = None
-    filename: str | None = None
+__all__ = ["RawMessage"]
diff --git a/syft/sync/sync/delivery_tracker.py b/syft/sync/sync/delivery_tracker.py
new file mode 100644
index 00000000000..963bdd8479a
--- /dev/null
+++ b/syft/sync/sync/delivery_tracker.py
@@ -0,0 +1,153 @@
+"""What each peer holds, so the data owner can resend, delete and catch up.
+
+STUB: the names and contracts are agreed; the bodies are to be written.
+
+The tracker knows nothing about the transport or where its record is kept. It
+works on paths and content hashes, takes the outcome of each send as a
+``SendStatus``, and keeps its record through a ``DeliveryStateStore``.
+
+Views
+-----
+For every peer the tracker keeps a view, ``{path: hash}``: the version of each
+file that peer is confirmed to hold. ``plan`` compares a view with what the
+peer may read now:
+
+    readable, and the hash differs from the view  -> send
+    in the view, but no longer readable           -> delete
+    in the view, but the file is gone             -> delete
+
+A peer with no view, because it is new or its record was lost, gets a
+``PeerReset``: its view starts empty, so it is sent every file it may read.
+
+Recording a send
+----------------
+``sent`` takes the ``SendStatus`` from ``P2PConnection.send``:
+
+- ``DELIVERED``: the view changes at once.
+- ``QUEUED``: the changes are held under their ``msg_id``, and the view changes
+  only when ``P2PConnection.delivered()`` reports that ``msg_id`` and
+  ``confirm`` is called. A receipt is not enough; only the peer's ack counts.
+
+A view therefore never changes before a delivery is confirmed.
+
+Pending deliveries
+------------------
+Changes queued for a peer but not yet sent, for example because the upload
+failed or the peer was skipped by the version gate, are kept per peer by the
+owner cache's pending deliveries (PR #9563). The tracker does not duplicate
+them.
+
+Storage
+-------
+``DeliveryState`` is the whole record, versioned. A ``DeliveryStateStore``
+loads and saves it: a local file next to the owner's file hashes, and with
+``persist_owner_state`` also Drive, inside the rolling state and checkpoints,
+so an enclave that wipes its local state at every boot does not reset every
+peer.
+
+Sends not routed by permission, such as the echo of a submitter's own change
+or a private dataset shared with an enclave, are recorded only where the
+recipient may read the file, so a later permission change never deletes them.
+
+Open decisions:
+- A peer rejected, then approved again: reset its view and resend (proposed).
+- Accepted job submissions: echo only to the submitter, as today (proposed), or
+  also send to every peer who may read the file.
+"""
+
+from abc import ABC, abstractmethod
+from typing import Iterable, Mapping
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from syft.sync.transport.types import SendStatus
+
+DELIVERY_VIEWS_FILENAME = "owner_delivery_views.json"
+# Format of DeliveryState. Raise it when the layout changes, and keep a read
+# path for every earlier version.
+DELIVERY_STATE_VERSION = 1
+
+
+class QueuedSend(BaseModel):
+    """Changes sent to ``peer`` that wait for confirmation.
+
+    ``changes`` maps each path to its new hash, or None for a delete.
+    """
+
+    peer: str
+    changes: dict[str, str | None]
+
+
+class DeliveryState(BaseModel):
+    """Everything the tracker keeps: views, and sends awaiting confirmation."""
+
+    version: int = DELIVERY_STATE_VERSION
+    # peer -> {path: hash}
+    views: dict[str, dict[str, str]] = Field(default_factory=dict)
+    # msg_id -> the send it confirms
+    queued: dict[str, QueuedSend] = Field(default_factory=dict)
+
+
+class DeliveryPlan(BaseModel):
+    """What ``plan`` decided for one peer."""
+
+    model_config = ConfigDict(frozen=True)
+
+    send: list[str] = Field(default_factory=list)
+    delete: list[str] = Field(default_factory=list)
+
+
+class DeliveryStateStore(ABC):
+    """Where a ``DeliveryState`` is kept: a local file, Drive, or memory."""
+
+    @abstractmethod
+    def load(self) -> DeliveryState:
+        """The stored state, or an empty one."""
+
+    @abstractmethod
+    def save(self, state: DeliveryState) -> None:
+        """Store ``state``, replacing what was there."""
+
+
+class DeliveryTracker:
+    """Per-peer views of confirmed file versions, for the data owner."""
+
+    def __init__(self, store: DeliveryStateStore):
+        self.store = store
+
+    def has_view(self, peer: str) -> bool:
+        """Whether ``peer`` has a view. False means a ``PeerReset`` is due."""
+        raise NotImplementedError
+
+    def reset(self, peer: str) -> None:
+        """Start ``peer``'s view empty (a ``PeerReset``), so the next plan sends
+        it every file it may read."""
+        raise NotImplementedError
+
+    def plan(
+        self, peer: str, readable: Mapping[str, str], paths: Iterable[str]
+    ) -> DeliveryPlan:
+        """What to send and delete for ``peer`` among ``paths``.
+
+        ``readable`` maps every path ``peer`` may read now to its current hash.
+        ``paths`` are the dirty paths; for a peer with no view, every readable
+        path.
+        """
+        raise NotImplementedError
+
+    def sent(
+        self,
+        peer: str,
+        msg_id: str,
+        status: SendStatus,
+        changes: Mapping[str, str | None],
+    ) -> None:
+        """Record a send of ``changes`` (path -> new hash, None for a delete).
+
+        ``DELIVERED`` updates the view now; ``QUEUED`` waits for ``confirm``.
+        """
+        raise NotImplementedError
+
+    def confirm(self, msg_ids: Iterable[str]) -> None:
+        """Apply the queued sends that ``P2PConnection.delivered()`` reported."""
+        raise NotImplementedError
diff --git a/syft/sync/transport/__init__.py b/syft/sync/transport/__init__.py
new file mode 100644
index 00000000000..f7460ff2e7e
--- /dev/null
+++ b/syft/sync/transport/__init__.py
@@ -0,0 +1,23 @@
+"""Backend-neutral transport contracts for the syft sync engine."""
+
+from syft.sync.transport.protocols import (
+    Collections,
+    OwnerStore,
+    P2PConnection,
+    Peering,
+    Transport,
+    WatcherStore,
+)
+from syft.sync.transport.types import MessageKind, RawMessage, SendStatus
+
+__all__ = [
+    "Collections",
+    "MessageKind",
+    "OwnerStore",
+    "P2PConnection",
+    "Peering",
+    "RawMessage",
+    "SendStatus",
+    "Transport",
+    "WatcherStore",
+]
diff --git a/syft/sync/transport/naming.py b/syft/sync/transport/naming.py
new file mode 100644
index 00000000000..87877b0efa0
--- /dev/null
+++ b/syft/sync/transport/naming.py
@@ -0,0 +1,35 @@
+"""Syft's naming rules for stored files and documents.
+
+These rules belong to syft, not to any backend: a backend stores names, and the
+sync layer gives them meaning. They move here from the Drive transport one at a
+time.
+"""
+
+EVENTS_MESSAGE_PREFIX = "syfteventsmessagev3_"
+PROPOSALS_MESSAGE_PREFIX = "msgv2_"
+
+# SYFT_peers.json is a flat map of peer email to entry, so a version at the top
+# level would look like a peer email. The version goes under this reserved key.
+# A client written before the key reads a peer state from that entry and fails.
+# The key therefore never appears as a peer.
+PEERS_META_KEY = "_meta"
+# Shape of one entry in SYFT_peers.json. Raise this when an entry changes. A file
+# with no reserved entry was written before the version, and is version 0.
+SYFT_PEERS_VERSION = 1
+
+
+def timestamp_from_message_name(name: str) -> float | None:
+    """The send time a message file carries in its name, or None.
+
+    Both message formats put the time second: ``syfteventsmessagev3_<ts>_<id>``
+    for events and ``msgv2_<ts>_<id>`` for proposals.
+    """
+    if not name.startswith((EVENTS_MESSAGE_PREFIX, PROPOSALS_MESSAGE_PREFIX)):
+        return None
+    parts = name.split("_")
+    if len(parts) < 2:
+        return None
+    try:
+        return float(parts[1])
+    except ValueError:
+        return None
diff --git a/syft/sync/transport/protocols.py b/syft/sync/transport/protocols.py
new file mode 100644
index 00000000000..f8b4ec2e62e
--- /dev/null
+++ b/syft/sync/transport/protocols.py
@@ -0,0 +1,310 @@
+"""The jobs a transport backend does, one abstract class each.
+
+``P2PConnection`` carries messages between peers, ``OwnerStore`` and
+``WatcherStore`` hold each role's own storage, ``Collections`` shares datasets,
+and ``Peering`` finds and trusts peers. A backend implements the classes it
+can. The sync layer above never sees backend ids: it passes back the messages
+and handles a backend returns.
+
+Each class keeps its abstract methods to the few a backend must write. Methods
+that only combine those are ordinary methods on the base class, which a
+backend may override when it can do them faster.
+"""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from typing import TYPE_CHECKING
+
+from pydantic import BaseModel, ConfigDict
+
+from syft.sync.platforms.base_platform import BasePlatform
+from syft.sync.transport.types import MessageKind, RawMessage, SendStatus
+
+if TYPE_CHECKING:
+    from syft.sync.connections.base_connection import FileCollection
+    from syft.sync.version.version_info import VersionInfo
+
+
+class P2PConnection(ABC):
+    """Messages between peers, kept until the recipient acknowledges them.
+
+    One rule for every message, whatever its kind: ``receive`` returns the
+    messages not acknowledged yet, and ``ack`` finishes one. On Drive a written
+    message is ``DELIVERED`` and ``ack`` removes it. On a live link such as
+    WebRTC a message is ``QUEUED`` until the peer acknowledges it, which
+    ``delivered()`` then reports.
+    """
+
+    @abstractmethod
+    def send(
+        self, peer: str, msg_id: str, data: bytes, kind: MessageKind
+    ) -> SendStatus:
+        """Send ``data`` to ``peer`` under ``msg_id``.
+
+        ``msg_id`` is chosen by the sender and stays the same if the message is
+        sent again. ``kind`` lets a connection keep the two kinds apart, as
+        Drive does with an inbox and an outbox.
+        """
+
+    @abstractmethod
+    def delivered(self) -> list[str]:
+        """The ``msg_id`` of each queued message the peer acknowledged since
+        the last call. A connection that only returns ``DELIVERED`` returns
+        nothing."""
+
+    @abstractmethod
+    def receive(self, max_messages: int | None = None) -> list[RawMessage]:
+        """Messages from every peer that are not acknowledged yet, oldest first
+        per sender, at most ``max_messages`` of them.
+
+        A peer whose channel does not exist yet has no messages. The connection
+        sets each ``RawMessage.sender`` from where the message came from.
+        """
+
+    @abstractmethod
+    def ack(self, message: RawMessage) -> None:
+        """Record that ``message`` was processed; ``receive`` never returns it
+        again."""
+
+
+class OwnerStore(ABC):
+    """The data owner's own storage: event log, checkpoints, rolling state.
+
+    Only the owner writes and reads it, and it is never shared.
+    """
+
+    # -- event log -------------------------------------------------------------
+    @abstractmethod
+    def append_event(self, name: str, data: bytes) -> str:
+        """Append one events message to the event log; return its handle."""
+
+    @abstractmethod
+    def list_event_handles(self, since_timestamp: float | None = None) -> list[str]:
+        """Handles of the events messages sent after ``since_timestamp``, oldest
+        first; every message when it is None."""
+
+    @abstractmethod
+    def fetch(self, handle: str) -> bytes:
+        """The bytes stored under ``handle``."""
+
+    def fetch_events_since(self, since_timestamp: float | None) -> list[bytes]:
+        """The events messages sent after ``since_timestamp``, oldest first."""
+        return [self.fetch(h) for h in self.list_event_handles(since_timestamp)]
+
+    def fetch_all_events(self) -> list[bytes]:
+        """Every events message in the event log, oldest first."""
+        return self.fetch_events_since(None)
+
+    def count_events_since(self, since_timestamp: float | None) -> int:
+        """How many events messages were sent after ``since_timestamp``."""
+        return len(self.list_event_handles(since_timestamp))
+
+    # -- checkpoints -----------------------------------------------------------
+    @abstractmethod
+    def put_checkpoint(self, name: str, data: bytes) -> str:
+        """Store a full checkpoint; it replaces every earlier one."""
+
+    @abstractmethod
+    def latest_checkpoint(self) -> bytes | None:
+        """The newest full checkpoint, or None."""
+
+    @abstractmethod
+    def put_incremental(self, name: str, data: bytes) -> str:
+        """Store an incremental checkpoint."""
+
+    @abstractmethod
+    def all_incrementals(self) -> list[bytes]:
+        """Every incremental checkpoint since the last full one."""
+
+    @abstractmethod
+    def incremental_count(self) -> int:
+        """How many incremental checkpoints are stored, without fetching them."""
+
+    @abstractmethod
+    def next_incremental_sequence(self) -> int:
+        """The sequence number the next incremental checkpoint takes."""
+
+    @abstractmethod
+    def delete_incrementals(self) -> None:
+        """Remove every incremental checkpoint."""
+
+    # -- rolling state ---------------------------------------------------------
+    @abstractmethod
+    def put_rolling_state(self, name: str, data: bytes) -> str:
+        """Store the rolling state, overwriting the previous one."""
+
+    @abstractmethod
+    def get_rolling_state(self) -> bytes | None:
+        """The rolling state, or None."""
+
+    @abstractmethod
+    def delete_rolling_state(self) -> None:
+        """Remove the rolling state."""
+
+
+class WatcherStore(ABC):
+    """The data scientist's memory of what each owner sent it.
+
+    One area per owner, in the scientist's own storage, never shared, so it
+    survives a lost local disk such as a Colab runtime reset.
+    """
+
+    @abstractmethod
+    def put_checkpoint(self, peer: str, name: str, data: bytes) -> str:
+        """Store a snapshot of every file received from ``peer``; it replaces
+        the previous one."""
+
+    @abstractmethod
+    def latest_checkpoint(self, peer: str) -> bytes | None:
+        """The newest snapshot of files received from ``peer``, or None."""
+
+    @abstractmethod
+    def put_rolling_state(self, peer: str, name: str, data: bytes) -> str:
+        """Store the changes received from ``peer`` since the snapshot, and the
+        ids of the messages already processed."""
+
+    @abstractmethod
+    def get_rolling_state(self, peer: str) -> bytes | None:
+        """The rolling state for ``peer``, or None."""
+
+    @abstractmethod
+    def reset(self, peer: str) -> None:
+        """Forget everything received from ``peer``, before asking it to send
+        everything again."""
+
+
+class Collections(ABC):
+    """Sets of files a data owner shares with named readers (datasets)."""
+
+    @abstractmethod
+    def create(self, prefix: str, tag: str, content_hash: str, owner_email: str) -> str:
+        """Create the collection if it does not exist; return its handle."""
+
+    @abstractmethod
+    def mark_shared_with_any(self, prefix: str, tag: str, content_hash: str) -> None:
+        """Flag the collection as meant for every approved peer."""
+
+    @abstractmethod
+    def grant_read(
+        self, prefix: str, tag: str, content_hash: str, users: list[str]
+    ) -> None:
+        """Let ``users`` read the collection. Granting twice is harmless."""
+
+    @abstractmethod
+    def upload(
+        self, prefix: str, tag: str, content_hash: str, files: dict[str, bytes]
+    ) -> None:
+        """Store ``files`` in the collection."""
+
+    @abstractmethod
+    def list_mine_with_permissions(self, prefix: str) -> list[FileCollection]:
+        """The collections I published under ``prefix``, with their flags."""
+
+    def list_mine(self, prefix: str) -> list[str]:
+        """The tags of the collections I published under ``prefix``, each once."""
+        tags = (c.tag for c in self.list_mine_with_permissions(prefix))
+        return list(dict.fromkeys(tags))
+
+    @abstractmethod
+    def delete(self, prefix: str, tag: str) -> None:
+        """Remove every layout of ``tag`` published under ``prefix``."""
+
+    @abstractmethod
+    def list_shared_with_me(self, prefix: str) -> list[dict]:
+        """Collections peers shared with me, each with owner_email, tag,
+        content_hash and variant."""
+
+    @abstractmethod
+    def list_files(
+        self, prefix: str, tag: str, content_hash: str, owner_email: str
+    ) -> list[dict]:
+        """The files of one shared collection, each with file_id and file_name."""
+
+    @abstractmethod
+    def fetch_file(self, handle: str) -> bytes:
+        """The bytes of one collection file."""
+
+    def download(
+        self, prefix: str, tag: str, content_hash: str, owner_email: str
+    ) -> dict[str, bytes]:
+        """Every file of one collection a peer shared with me, by file name."""
+        files = self.list_files(prefix, tag, content_hash, owner_email)
+        return {f["file_name"]: self.fetch_file(f["file_id"]) for f in files}
+
+
+class Peering(ABC):
+    """Finding and trusting a peer: identity, channels, peer list, published
+    documents."""
+
+    @abstractmethod
+    def whoami(self) -> str:
+        """The email of the account behind this backend's credentials."""
+
+    @abstractmethod
+    def open_channels(self, peer: str) -> None:
+        """Create the message channels with ``peer`` and let it write to them."""
+
+    @abstractmethod
+    def incoming_requests(self) -> list[str]:
+        """Emails of peers who opened channels towards me."""
+
+    # -- peer list (SYFT_peers.json) -------------------------------------------
+    @abstractmethod
+    def read_peer_registry(self, force_download: bool = False) -> dict:
+        """My record of every peer and its state, keyed by email."""
+
+    @abstractmethod
+    def update_peer_registry(
+        self, peer: str, state: str, public_encryption_bundle: dict | None = None
+    ) -> None:
+        """Record ``peer``'s state, and its key bundle when given."""
+
+    # -- key bundles -----------------------------------------------------------
+    @abstractmethod
+    def publish_key_bundle(self, peer: str, bundle_json: str) -> None:
+        """Publish my public key bundle for ``peer``."""
+
+    @abstractmethod
+    def grant_key_bundle_read(self, peer: str) -> None:
+        """Let ``peer`` read the key bundles I publish."""
+
+    @abstractmethod
+    def read_own_key_bundle(self, peer: str) -> str | None:
+        """The key bundle I published for ``peer``, or None."""
+
+    @abstractmethod
+    def read_peer_key_bundle(self, peer: str) -> str | None:
+        """The key bundle ``peer`` published for me, or None."""
+
+    # -- version documents -----------------------------------------------------
+    @abstractmethod
+    def publish_version(self, version_info: VersionInfo) -> None:
+        """Publish my version document."""
+
+    @abstractmethod
+    def read_own_version(self) -> VersionInfo | None:
+        """My published version document, or None."""
+
+    @abstractmethod
+    def read_peer_version(self, peer: str) -> VersionInfo | None:
+        """``peer``'s published version document, or None."""
+
+    @abstractmethod
+    def grant_version_read(self, peer: str) -> None:
+        """Let ``peer`` read my version document."""
+
+
+class Transport(BaseModel):
+    """One implementation of each job. The parts may come from different
+    backends, though a peer's ``p2p`` and ``peering`` go together."""
+
+    model_config = ConfigDict(frozen=True, arbitrary_types_allowed=True)
+
+    p2p: P2PConnection
+    owner_store: OwnerStore
+    watcher_store: WatcherStore
+    collections: Collections
+    peering: Peering
+    # Recorded on each Peer so a peer list says which backend reaches it.
+    platform: BasePlatform
diff --git a/syft/sync/transport/types.py b/syft/sync/transport/types.py
new file mode 100644
index 00000000000..1f9cffa0d36
--- /dev/null
+++ b/syft/sync/transport/types.py
@@ -0,0 +1,40 @@
+"""Types shared by every transport backend."""
+
+from enum import Enum
+
+from pydantic import BaseModel, ConfigDict
+
+
+class MessageKind(str, Enum):
+    """What a message carries. Both kinds can travel between the same two peers."""
+
+    # Data scientist -> data owner: proposed file changes (job submissions).
+    PROPOSED_CHANGES = "proposed_changes"
+    # Data owner -> data scientist: accepted file change events (results).
+    FILE_EVENTS = "file_events"
+
+
+class SendStatus(str, Enum):
+    """What ``P2PConnection.send`` achieved."""
+
+    # Stored where the peer will find it, even while it is offline.
+    DELIVERED = "delivered"
+    # Not stored yet; confirmed later through ``P2PConnection.delivered()``.
+    QUEUED = "queued"
+
+
+class RawMessage(BaseModel):
+    """One received message, as bytes, before decoding or decryption.
+
+    ``sender`` is set by the connection from where the message came from,
+    never from its contents. ``handle`` is opaque: only the connection that
+    returned the message reads it, to acknowledge it.
+    """
+
+    model_config = ConfigDict(frozen=True)
+
+    sender: str
+    msg_id: str
+    kind: MessageKind
+    data: bytes
+    handle: str | None = None
diff --git a/tests/unit/test_transport_contracts.py b/tests/unit/test_transport_contracts.py
new file mode 100644
index 00000000000..dbb84239d10
--- /dev/null
+++ b/tests/unit/test_transport_contracts.py
@@ -0,0 +1,140 @@
+"""The transport contracts: the abstract jobs, their helper methods, the
+message types, the naming rules and the DeliveryTracker stub. No backend
+implements them yet."""
+
+from types import SimpleNamespace
+from unittest.mock import Mock
+
+import pytest
+from pydantic import ValidationError
+
+from syft.sync.platforms.gdrive_files_platform import GdriveFilesPlatform
+from syft.sync.sync.delivery_tracker import (
+    DELIVERY_STATE_VERSION,
+    DeliveryState,
+    DeliveryStateStore,
+    DeliveryTracker,
+)
+from syft.sync.transport import (
+    Collections,
+    MessageKind,
+    OwnerStore,
+    P2PConnection,
+    Peering,
+    RawMessage,
+    Transport,
+    WatcherStore,
+)
+from syft.sync.transport.naming import timestamp_from_message_name
+
+JOBS = [P2PConnection, OwnerStore, WatcherStore, Collections, Peering]
+
+
+@pytest.mark.parametrize("job", JOBS)
+def test_a_job_cannot_be_used_without_a_backend(job):
+    with pytest.raises(TypeError):
+        job()
+
+
+def test_owner_store_helpers_are_built_from_list_and_fetch():
+    store = Mock(spec=OwnerStore)
+    store.list_event_handles.return_value = ["h1", "h2"]
+    store.fetch.side_effect = lambda handle: f"bytes of {handle}".encode()
+
+    assert OwnerStore.fetch_events_since(store, 5.0) == [b"bytes of h1", b"bytes of h2"]
+    store.list_event_handles.assert_called_with(5.0)
+    assert OwnerStore.count_events_since(store, 5.0) == 2
+
+
+def test_fetching_all_events_lists_from_the_beginning():
+    store = Mock(spec=OwnerStore)
+    store.fetch_events_since.return_value = [b"all"]
+    assert OwnerStore.fetch_all_events(store) == [b"all"]
+    store.fetch_events_since.assert_called_once_with(None)
+
+
+def test_list_mine_returns_each_tag_once_in_order():
+    collections = Mock(spec=Collections)
+    collections.list_mine_with_permissions.return_value = [
+        SimpleNamespace(tag="census"),
+        SimpleNamespace(tag="census"),  # the same dataset in a second layout
+        SimpleNamespace(tag="sales"),
+    ]
+    assert Collections.list_mine(collections, "syft_datasetcollection") == [
+        "census",
+        "sales",
+    ]
+
+
+def test_download_fetches_every_listed_file_by_name():
+    collections = Mock(spec=Collections)
+    collections.list_files.return_value = [
+        {"file_id": "id-1", "file_name": "mock.csv"},
+        {"file_id": "id-2", "file_name": "readme.md"},
+    ]
+    collections.fetch_file.side_effect = lambda handle: handle.encode()
+
+    files = Collections.download(collections, "prefix", "census", "abc", "do@x.org")
+    assert files == {"mock.csv": b"id-1", "readme.md": b"id-2"}
+
+
+def test_a_transport_bundles_one_of_each_job_and_cannot_be_changed():
+    transport = Transport(
+        p2p=Mock(spec=P2PConnection),
+        owner_store=Mock(spec=OwnerStore),
+        watcher_store=Mock(spec=WatcherStore),
+        collections=Mock(spec=Collections),
+        peering=Mock(spec=Peering),
+        platform=GdriveFilesPlatform(),
+    )
+    with pytest.raises(ValidationError):
+        transport.p2p = Mock(spec=P2PConnection)
+
+
+def test_a_received_message_cannot_be_changed():
+    message = RawMessage(
+        sender="ds@example.com",
+        msg_id="msgv2_1.0_abc",
+        kind=MessageKind.PROPOSED_CHANGES,
+        data=b"job",
+        handle="file-id",
+    )
+    with pytest.raises(ValidationError):
+        message.sender = "someone-else@example.com"
+
+
+def test_the_old_raw_message_import_path_still_works():
+    from syft.sync.messages.raw_message import RawMessage as OldPathRawMessage
+
+    assert OldPathRawMessage is RawMessage
+
+
+@pytest.mark.parametrize(
+    "name, expected",
+    [
+        ("syfteventsmessagev3_1759800000.95_abc.tar.gz", 1759800000.95),
+        ("msgv2_12.5_abc.tar.gz", 12.5),
+        ("checkpoint_12.5.tar.gz", None),
+        ("msgv2_notanumber_abc.tar.gz", None),
+    ],
+)
+def test_message_names_carry_their_send_time(name, expected):
+    assert timestamp_from_message_name(name) == expected
+
+
+def test_delivery_tracker_is_a_stub_until_implemented():
+    tracker = DeliveryTracker(store=Mock(spec=DeliveryStateStore))
+    with pytest.raises(NotImplementedError):
+        tracker.has_view("ds@example.com")
+
+
+def test_a_delivery_state_store_needs_a_backend():
+    with pytest.raises(TypeError):
+        DeliveryStateStore()
+
+
+def test_an_empty_delivery_state_is_versioned():
+    state = DeliveryState()
+    assert state.version == DELIVERY_STATE_VERSION
+    assert state.views == {}
+    assert state.queued == {}