feat: versioned objects and migrations for syft-client

packages/syft-datasets/src/syft_datasets/dataset_manager.py

@@ -1,7 +1,9 @@
 from pathlib import Path
+
 from typing_extensions import Self
 
 import yaml
+from syft_migration import ProtocolSchema
 
 from .types import PathLike, to_path
 from syft_notebook_ui.types import TableList
@@ -20,18 +22,34 @@
 
 
 class SyftDatasetManager:
-    def __init__(self, syftbox_folder_path: PathLike, email: str):
+    def __init__(
+        self,
+        syftbox_folder_path: PathLike,
+        email: str,
+        peer_schemas: dict[str, ProtocolSchema] | None = None,
+    ):
         self.syftbox_config = SyftBoxConfig(
             syftbox_folder=to_path(syftbox_folder_path), email=email
         )
-        # peer_schemas (peer email -> dataset ProtocolSchema) will be filled in by
-        # syft-client later; until then every peer resolves to the widest-
-        # compatible protocol, so datasets are written in that layout.
-        self.storage = DatasetStorage(config=self.syftbox_config)
+        # peer_schemas (peer email -> dataset ProtocolSchema): syft-client
+        # passes PeerManager's live map here (updated in place as peer version
+        # files load). Peers without an entry resolve to the widest-compatible
+        # protocol, so datasets stay readable by unknown-version peers.
+        self.storage = DatasetStorage(
+            config=self.syftbox_config, peer_schemas=peer_schemas
+        )
 
     @classmethod
-    def from_config(cls, config: SyftBoxConfig) -> Self:
-        return cls(syftbox_folder_path=config.syftbox_folder, email=config.email)
+    def from_config(
+        cls,
+        config: SyftBoxConfig,
+        peer_schemas: dict[str, ProtocolSchema] | None = None,
+    ) -> Self:
+        return cls(
+            syftbox_folder_path=config.syftbox_folder,
+            email=config.email,
+            peer_schemas=peer_schemas,
+        )
 
     def create(
         self,

packages/syft-datasets/src/syft_datasets/dataset_storage.py

@@ -98,10 +98,15 @@ def __init__(
         self.config = config
         self.registry = registry
         self.service = MigrationService(registry=registry)
-        # peer email -> dataset ProtocolSchema; filled in by syft-client later.
-        # Peers without an entry cannot be assumed to read the current layout, so
+        # peer email -> dataset ProtocolSchema; syft-client passes PeerManager's
+        # live map here (updated in place as peer version files load). Peers
+        # without an entry cannot be assumed to read the current layout, so
         # they resolve to the widest-compatible (oldest) protocol.
-        self.peer_schemas: dict[str, ProtocolSchema] = peer_schemas or {}
+        # `is not None`, not `or`: the live dict starts empty and `or {}` would
+        # drop the shared reference, freezing negotiation at construction time.
+        self.peer_schemas: dict[str, ProtocolSchema] = (
+            peer_schemas if peer_schemas is not None else {}
+        )
         self.codecs = [cls(config) for cls in CODECS]
 
     @property

packages/syft-job/src/syft_job/client.py

@@ -81,8 +81,12 @@ def __init__(
         self._validate_user_email()
 
     @classmethod
-    def from_config(cls, config: SyftJobConfig) -> "JobClient":
-        return cls(config, config.current_user_email)
+    def from_config(
+        cls,
+        config: SyftJobConfig,
+        peer_schemas: Optional[dict[str, ProtocolSchema]] = None,
+    ) -> "JobClient":
+        return cls(config, config.current_user_email, peer_schemas=peer_schemas)
 
     def _validate_user_email(self) -> None:
         """Validate that the user_email directory exists in SyftBox root."""

packages/syft-job/src/syft_job/job_storage.py

@@ -37,9 +37,15 @@ def __init__(
         self.config = config
         self.registry = registry
         self.service = MigrationService(registry=registry)
-        # peer email -> job ProtocolSchema; filled in by syft-client later.
-        # Peers without an entry are assumed to run the current protocol.
-        self.peer_schemas: dict[str, ProtocolSchema] = peer_schemas or {}
+        # peer email -> job ProtocolSchema; syft-client passes PeerManager's
+        # live map here (updated in place as peer version files load). Peers
+        # without an entry are assumed to run the current protocol.
+        # `is not None`, not `or`: syft-client passes a live (initially empty)
+        # dict it mutates as peer version files load; `or {}` would drop the
+        # shared reference and freeze negotiation at construction-time state.
+        self.peer_schemas: dict[str, ProtocolSchema] = (
+            peer_schemas if peer_schemas is not None else {}
+        )
         self.codecs = [cls(config) for cls in CODECS]
 
     @property
@@ -106,8 +112,8 @@ def new_submission_ref(self, do_email: str, job_name: str) -> JobRef:
             datasite_email=do_email,
             ds_email=self.config.current_user_email,
             job_name=job_name,
-            # Until syft-client fills peer_schemas, unknown peers are assumed
-            # to run the current protocol.
+            # Peers without a known schema are assumed to run the current
+            # protocol.
             protocol_version=self.negotiated_protocol_version_for_peer(
                 do_email, raise_on_unknown=False
             ),

pyproject.toml

@@ -42,6 +42,7 @@ dependencies = [
     "syft-dataset==0.1.20", # TODO: move to optional-dependencies after adding lazy imports
     "syft-permissions==0.1.14",
     "syft-perms==0.1.14",
+    "syft-migration",
     "syft-crypto-python>=0.1.2b2",
 ]
 

scripts/export_release_artifact.py

@@ -0,0 +1,58 @@
+"""Export the release artifacts for the current syft-client version.
+
+Run on EVERY release (uv run python scripts/export_release_artifact.py):
+always writes the package release info; additionally writes the protocol
+artifact when this release introduces a new protocol version.
+"""
+
+import sys
+
+from syft_client.migrations.history import PACKAGE_ARTIFACTS_DIR, PROTOCOLS_DIR
+from syft_client.migrations.registry import (
+    SYFT_CLIENT_PROTOCOL_VERSION,
+    client_registry,
+)
+from syft_client.version import SYFT_CLIENT_VERSION
+
+
+def main() -> None:
+    # Import the package so every versioned object is registered.
+    import syft_client  # noqa: F401
+
+    if client_registry.protocol_changed_without_bump():
+        sys.exit(
+            "The syft-client protocol changed compared to the released "
+            f"protocol-{SYFT_CLIENT_PROTOCOL_VERSION}.json; bump "
+            "SYFT_CLIENT_PROTOCOL_VERSION in syft_client/migrations/registry.py "
+            "before releasing."
+        )
+
+    info_path = PACKAGE_ARTIFACTS_DIR / f"syft-client-{SYFT_CLIENT_VERSION}.json"
+    protocol_path = PROTOCOLS_DIR / f"protocol-{SYFT_CLIENT_PROTOCOL_VERSION}.json"
+    need_info = not info_path.exists()
+    need_protocol = not protocol_path.exists()
+
+    # Single exit when there is nothing left to write
+    if not need_info and not need_protocol:
+        sys.exit(
+            f"Release artifacts already present:\n"
+            f"  {info_path}\n"
+            f"  {protocol_path}\n"
+            "They are frozen once written. Bump SYFT_CLIENT_VERSION (and "
+            "SYFT_CLIENT_PROTOCOL_VERSION if the protocol changed) before "
+            "exporting again."
+        )
+
+    if need_info:
+        client_registry.compute_released_package_protocol_info().save(info_path)
+        print(f"Wrote {info_path}")
+    else:
+        print(f"Package artifact already present: {info_path}")
+
+    if need_protocol:
+        client_registry.compute_released_protocol().save(protocol_path)
+        print(f"Wrote {protocol_path} (new protocol version)")
+
+
+if __name__ == "__main__":
+    main()

scripts/generate_release_fixture.py

@@ -0,0 +1,119 @@
+"""Generate a p2p backward-compatibility fixture for the current syft-client release.
+
+Run on EVERY release, after bumping the version:
+
+    uv run python scripts/generate_release_fixture.py
+
+Writes the serialized artifacts exactly as this release produces them, into
+
+    tests/migrations/p2p/fixtures/syft_client-<version>-protocol<p>/
+        SYFT_version.json                       # the published version file
+        msgv2_<...>.tar.gz                      # a proposed-changes message (DS -> DO)
+        syfteventsmessagev3_<...>.tar.gz        # an events message (DO -> watchers)
+
+Unlike syft-job there is no local SyftBox tree to snapshot (storage is the
+Google Drive transport), so fixtures are directories of captured blobs; future
+releases loop over them (test_older_protocol_compatibility.py) to prove they
+can still read and round-trip older serialized data.
+
+Protocol 0 / release 0.1.117 predates this script; its fixture
+(syft_client-0.1.117-protocol0) is hand-authored, like protocol-0.json.
+"""
+
+import sys
+from pathlib import Path
+
+from syft_client.migrations.registry import SYFT_CLIENT_PROTOCOL_VERSION
+from syft_client.sync.events.file_change_event import (
+    FileChangeEvent,
+    FileChangeEventsMessage,
+)
+from syft_client.sync.messages.proposed_filechange import (
+    ProposedFileChange,
+    ProposedFileChangesMessage,
+)
+from syft_client.sync.version.version_info import VersionInfo
+from syft_client.version import SYFT_CLIENT_VERSION
+
+DO_EMAIL = "do@test.org"
+DS_EMAIL = "ds@test.org"
+
+FIXTURES_DIR = (
+    Path(__file__).resolve().parents[1] / "tests" / "migrations" / "p2p" / "fixtures"
+)
+
+
+def build_version_info() -> VersionInfo:
+    # Not VersionInfo.current(): the detected install source is an absolute
+    # local path on dev machines, which must not leak into a committed fixture.
+    return VersionInfo.current().model_copy(
+        update={"syft_client_install_source": "pip"}
+    )
+
+
+def build_proposed_message() -> ProposedFileChangesMessage:
+    return ProposedFileChangesMessage(
+        sender_email=DS_EMAIL,
+        proposed_file_changes=[
+            ProposedFileChange(
+                path_in_datasite="data/notes.txt",
+                content="hello from the release fixture",
+                datasite_email=DO_EMAIL,
+            ),
+            ProposedFileChange(
+                path_in_datasite="data/blob.bin",
+                content=b"\x00\x01\x02fixture-binary",
+                datasite_email=DO_EMAIL,
+            ),
+            ProposedFileChange(
+                path_in_datasite="data/removed.txt",
+                content=None,
+                old_hash="0" * 64,
+                is_deleted=True,
+                datasite_email=DO_EMAIL,
+            ),
+        ],
+    )
+
+
+def build_events_message(
+    proposed: ProposedFileChangesMessage,
+) -> FileChangeEventsMessage:
+    events = [
+        FileChangeEvent.from_proposed_filechange(change)
+        for change in proposed.proposed_file_changes
+    ]
+    return FileChangeEventsMessage(events=events)
+
+
+def main() -> None:
+    # Any fixture for this version (any protocol) means the version was
+    # already released; a released version's serialized form is frozen.
+    existing = sorted(FIXTURES_DIR.glob(f"syft_client-{SYFT_CLIENT_VERSION}-protocol*"))
+    if existing:
+        sys.exit(
+            f"{existing[0]} already exists — fixtures are frozen once written. "
+            "Bump SYFT_CLIENT_VERSION before generating."
+        )
+    target = FIXTURES_DIR / (
+        f"syft_client-{SYFT_CLIENT_VERSION}-protocol{SYFT_CLIENT_PROTOCOL_VERSION}"
+    )
+    target.mkdir(parents=True)
+
+    (target / "SYFT_version.json").write_text(build_version_info().to_json())
+
+    proposed = build_proposed_message()
+    (target / proposed.message_filename.as_string()).write_bytes(
+        proposed.as_compressed_data()
+    )
+
+    events = build_events_message(proposed)
+    (target / events.message_filepath.as_string()).write_bytes(
+        events.as_compressed_data()
+    )
+
+    print(f"Wrote {target}")
+
+
+if __name__ == "__main__":
+    main()

syft_client/__init__.py

@@ -35,6 +35,15 @@
     delete_syftbox,
     delete_local_syftbox,
 )
+from syft_client.migrations.history import register_historic_schemas  # noqa: E402
+
+# Import the versioned model modules explicitly so registration is intentional,
+# not a side-effect of whatever login happened to pull in first.
+import syft_client.sync.version.version_info  # noqa: F401, E402
+import syft_client.sync.messages.proposed_filechange  # noqa: F401, E402
+import syft_client.sync.events.file_change_event  # noqa: F401, E402
+
+register_historic_schemas()
 
 SYFT_CLIENT_DIR = Path(__file__).parent.parent
 CREDENTIALS_DIR = SYFT_CLIENT_DIR / "credentials"

syft_client/migrations/__init__.py

@@ -0,0 +1,15 @@
+from .registry import (
+    PROTOCOL_NAME,
+    SYFT_CLIENT_PROTOCOL_VERSION,
+    client_migration_service,
+    client_registry,
+    load_as_latest,
+)
+
+__all__ = [
+    "PROTOCOL_NAME",
+    "SYFT_CLIENT_PROTOCOL_VERSION",
+    "client_migration_service",
+    "client_registry",
+    "load_as_latest",
+]

syft_client/migrations/history.py

@@ -0,0 +1,32 @@
+from pathlib import Path
+
+from syft_migration import ReleasedPackageProtocolInfo, ReleasedProtocol
+
+from .registry import client_registry
+
+# Release artifacts of past syft-client releases:
+#   package-artifacts/syft-client-<package_version>.json  (every release)
+#   protocols/protocol-<protocol_version>.json            (only when the protocol changed)
+# Generated by scripts/export_release_artifact.py; 0.1.117 / protocol 0 predate
+# the artifact mechanism, so their files are hardcoded as if that release had
+# emitted them.
+HISTORY_DIR = Path(__file__).parent / "history"
+PACKAGE_ARTIFACTS_DIR = HISTORY_DIR / "package-artifacts"
+PROTOCOLS_DIR = HISTORY_DIR / "protocols"
+
+
+def register_historic_schemas() -> None:
+    """Register the release artifacts of past releases into the client registry.
+
+    Must run after the versioned models are imported: with
+    ``raise_for_unknown_objects`` an artifact listing an object version this
+    release cannot load fails at import time instead of at migration time.
+    """
+    for path in sorted(PACKAGE_ARTIFACTS_DIR.glob("*.json")):
+        client_registry.register_released_package_protocol_info(
+            ReleasedPackageProtocolInfo.load(path), raise_for_unknown_objects=True
+        )
+    for path in sorted(PROTOCOLS_DIR.glob("*.json")):
+        client_registry.register_released_protocol(
+            ReleasedProtocol.load(path), raise_for_unknown_objects=True
+        )