feat: add versioned job objects with syft-migration registry

packages/syft-bg/src/syft_bg/notify/monitors/job.py

@@ -4,8 +4,7 @@
 from typing import Optional
 
 from syft_job.config import SyftJobConfig
-from syft_job.models.config import JobSubmissionMetadata
-from syft_job.models.state import JobState, JobStatus
+from syft_job.models import JobState, JobStatus, JobSubmissionMetadata
 
 from syft_bg.common.monitor import Monitor
 from syft_bg.common.state import JsonStateManager

packages/syft-enclave/src/syft_enclaves/client.py

@@ -8,8 +8,7 @@
 from syft_client.sync.peers.peer_list import PeerList
 from syft_datasets.dataset_manager import SyftDatasetManager
 from syft_job.job import JobInfo, JobsList
-from syft_job.models.config import JobSubmissionMetadata
-from syft_job.models.state import JobState, JobStatus
+from syft_job.models import JobState, JobStatus, JobSubmissionMetadata
 
 from syft_enclaves.enclave_job_info import (
     EnclaveJobInfo,

packages/syft-enclave/src/syft_enclaves/enclave_job_client.py

@@ -3,7 +3,7 @@
 
 from syft_job.client import BaseJobClient, JobClient
 from syft_job.job import JobsList
-from syft_job.models.config import JobSubmissionMetadata
+from syft_job.models import JobSubmissionMetadata
 
 
 class EnclaveJobClient(BaseJobClient):

packages/syft-enclave/src/syft_enclaves/enclave_job_info.py

@@ -7,7 +7,7 @@
 
 from pydantic import BaseModel
 from syft_job.job import JobInfo
-from syft_job.models.state import JobStatus
+from syft_job.models import JobStatus
 
 
 class PartyApprovalStatus(BaseModel):

packages/syft-job/pyproject.toml

@@ -10,6 +10,7 @@ dependencies = [
     "pandas",
     "psutil",
     "syft-perms==0.1.14",
+    "syft-migration",
 ]
 
 [project.scripts]

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

@@ -1,13 +1,15 @@
-__version__ = "0.1.25"
+# __version__ comes from the installed distribution metadata (see version.py).
+from .version import __version__
 
 from .client import BaseJobClient, JobClient, get_client
 from .config import SyftJobConfig
 from .job import JobInfo, JobsList
 from .job_runner import SyftJobRunner, create_runner
-from .models.config import JobSubmissionMetadata
-from .models.state import JobState, JobStatus
+from .migrations import job_registry
+from .models import JobState, JobStatus, JobSubmissionMetadata
 
 __all__ = [
+    "__version__",
     # SyftBox job system
     "BaseJobClient",
     "JobClient",
@@ -22,4 +24,6 @@
     "JobSubmissionMetadata",
     "JobState",
     "JobStatus",
+    # Migration registry
+    "job_registry",
 ]

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

@@ -12,8 +12,7 @@
 from .config import SyftJobConfig
 from .install_source import get_syft_client_install_source
 from .job import JobInfo, JobsList
-from .models.config import JobSubmissionMetadata
-from .models.state import JobState, JobStatus
+from .models import JobState, JobStatus, JobSubmissionMetadata
 
 # Python version used when creating virtual environments for job execution
 RUN_SCRIPT_PYTHON_VERSION = "3.12"

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

@@ -14,8 +14,7 @@
     jobs_list_str,
 )
 from .job_stdout import StdoutViewer
-from .models.config import JobSubmissionMetadata
-from .models.state import JobState, JobStatus
+from .models import JobState, JobStatus, JobSubmissionMetadata
 
 if TYPE_CHECKING:
     from .client import JobClient

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

@@ -12,8 +12,7 @@
 from .job import JobInfo
 from . import __version__
 from .config import SyftJobConfig
-from .models.state import JobState, JobStatus
-from .models.config import JobSubmissionMetadata
+from .models import JobState, JobStatus, JobSubmissionMetadata
 
 # Default timeout for job execution (10 minutes)
 DEFAULT_JOB_TIMEOUT_SECONDS = 600

packages/syft-job/src/syft_job/migrations/__init__.py

@@ -0,0 +1,5 @@
+from .registry import job_registry
+
+__all__ = [
+    "job_registry",
+]

packages/syft-job/src/syft_job/migrations/registry.py

@@ -0,0 +1,15 @@
+from syft_migration import MigrationRegistry
+
+from ..version import PACKAGE_NAME, __version__
+
+# Hardcoded, language-agnostic identifier for the syft-job protocol;
+# intentionally distinct from the package name.
+PROTOCOL_NAME = "syft-job"
+
+# Package-local registry for all versioned syft-job objects. The current
+# protocol schema is computed from the objects registered into it.
+job_registry = MigrationRegistry(
+    protocol_name=PROTOCOL_NAME,
+    package_name=PACKAGE_NAME,
+    package_version=__version__,
+)

packages/syft-job/src/syft_job/models/__init__.py

@@ -1,8 +1,12 @@
-from .config import JobSubmissionMetadata
-from .state import JobState, JobStatus
+from .job_state import JobState, JobStateV1, JobStatus
+from .job_submission_metadata import JobSubmissionMetadata, JobSubmissionMetadataV1
 
 __all__ = [
+    # Current-version aliases
     "JobSubmissionMetadata",
     "JobState",
     "JobStatus",
+    # Versioned objects
+    "JobSubmissionMetadataV1",
+    "JobStateV1",
 ]

packages/syft-job/src/syft_job/models/job_state/__init__.py

@@ -0,0 +1,10 @@
+from .v1 import JobStateV1, JobStatus
+
+# The current version of the job state object.
+JobState = JobStateV1
+
+__all__ = [
+    "JobState",
+    "JobStateV1",
+    "JobStatus",
+]

packages/syft-job/src/syft_job/models/state.py → packages/syft-job/src/syft_job/models/job_state/v1.py

@@ -6,7 +6,9 @@
 from typing import Optional
 
 import yaml
-from pydantic import BaseModel
+from syft_migration import MigratableObject
+
+from ...migrations import job_registry
 
 
 class JobStatus(str, Enum):
@@ -21,9 +23,12 @@ class JobStatus(str, Enum):
     FAILED = "failed"  # execution failed
 
 
-class JobState(BaseModel):
+class JobStateV1(MigratableObject, registry=job_registry):
     """Represents the state of a job, stored as state.yaml in the review/ directory."""
 
+    canonical_name: str = "JobState"
+    version: str = "1"
+
     status: JobStatus = JobStatus.RECEIVED
     received_at: Optional[datetime] = None
 
@@ -50,7 +55,7 @@ def save(self, path: Path) -> None:
             yaml.dump(self.model_dump(mode="json"), f, default_flow_style=False)
 
     @classmethod
-    def load(cls, path: Path) -> JobState:
+    def load(cls, path: Path) -> JobStateV1:
         """Load state from a YAML file."""
         with open(path, "r") as f:
             data = yaml.safe_load(f)

packages/syft-job/src/syft_job/models/job_submission_metadata/__init__.py

@@ -0,0 +1,9 @@
+from .v1 import JobSubmissionMetadataV1
+
+# The current version of the job submission metadata object.
+JobSubmissionMetadata = JobSubmissionMetadataV1
+
+__all__ = [
+    "JobSubmissionMetadata",
+    "JobSubmissionMetadataV1",
+]

packages/syft-job/src/syft_job/models/config.py → packages/syft-job/src/syft_job/models/job_submission_metadata/v1.py

@@ -6,13 +6,18 @@
 from typing import Literal, Optional
 
 import yaml
-from pydantic import BaseModel
+from syft_migration import MigratableObject
 
+from ...migrations import job_registry
 
-class JobSubmissionMetadata(BaseModel):
+
+class JobSubmissionMetadataV1(MigratableObject, registry=job_registry):
     """Represents the job submission metadata, stored under
     SyftBox/<datasite_email>/app_data/job/inbox/<ds_email>/<job_name>/config.yaml."""
 
+    canonical_name: str = "JobSubmissionMetadata"
+    version: str = "1"
+
     name: str
     type: Literal["python", "bash"] = "python"
     submitted_by: str
@@ -44,7 +49,7 @@ def is_valid_email(email):
         return re.match(r"^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+$", email) is not None
 
     @classmethod
-    def load(cls, path: Path) -> JobSubmissionMetadata:
+    def load(cls, path: Path) -> JobSubmissionMetadataV1:
         """Load config from a YAML file."""
         submitted_by = path.parent.parent.name
         datasite_email = path.parent.parent.parent.parent.parent.parent.name

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

@@ -0,0 +1,8 @@
+from importlib.metadata import version
+
+# Distribution name, computed from the import package name ("syft_job" -> "syft-job").
+PACKAGE_NAME = __package__.replace("_", "-")
+
+# Derived from the installed distribution metadata (pyproject.toml's version)
+# so it cannot drift from the released package version.
+__version__ = version(PACKAGE_NAME)

packages/syft-job/tests/migrations/fixtures/JobState/v1.yaml

@@ -0,0 +1,12 @@
+approval_method: null
+approved_at: null
+approved_by: null
+canonical_name: JobState
+completed_at: null
+received_at: null
+rejected_at: null
+rejected_by: null
+return_code: 0
+review_reason: null
+status: done
+version: '1'

packages/syft-job/tests/migrations/fixtures/JobSubmissionMetadata/v1.yaml

@@ -0,0 +1,17 @@
+canonical_name: JobSubmissionMetadata
+code_path: null
+datasets: null
+datasite_email: do@test.org
+dependencies:
+  - pandas
+entrypoint: main.py
+files:
+  - main.py
+headers: {}
+is_folder_submission: false
+job_type: local
+name: my.job
+submitted_at: '2026-07-01T00:00:00Z'
+submitted_by: ds@test.org
+type: python
+version: '1'

packages/syft-job/tests/migrations/mocks.py

@@ -0,0 +1,36 @@
+"""Shared mock objects for the syft-job migration tests."""
+
+from datetime import datetime, timezone
+from pathlib import Path
+
+from syft_job.models import JobSubmissionMetadataV1
+
+DO_EMAIL = "do@test.org"
+DS_EMAIL = "ds@test.org"
+
+
+def create_mock_submission() -> JobSubmissionMetadataV1:
+    return JobSubmissionMetadataV1(
+        name="my.job",
+        submitted_by=DS_EMAIL,
+        datasite_email=DO_EMAIL,
+        submitted_at=datetime(2026, 7, 1, tzinfo=timezone.utc),
+        entrypoint="main.py",
+        dependencies=["pandas"],
+        files=["main.py"],
+    )
+
+
+def mock_submission_config_path(tmp_path: Path) -> Path:
+    # config.yaml lives at inbox/<ds>/<job>/config.yaml under a datasite-email folder,
+    # matching the path layout JobSubmissionMetadataV1.load() reverse-engineers.
+    return (
+        tmp_path
+        / DO_EMAIL
+        / "app_data"
+        / "job"
+        / "inbox"
+        / DS_EMAIL
+        / "my.job"
+        / "config.yaml"
+    )

packages/syft-job/tests/migrations/test_migrations.py

@@ -0,0 +1,85 @@
+"""Migrations against the REAL job registry, using on-disk fixtures.
+
+Every registered version of each object has a serialized fixture under
+``fixtures/<canonical_name>/v<version>.yaml``; each fixture loads and upgrades
+in memory to the latest version, and the latest version downgrades to every
+registered version and writes to disk. With only V1 registered these are
+no-ops; when a newer version is registered, these tests fail until a fixture
+for it is added and then exercise the real migration paths automatically.
+"""
+
+from pathlib import Path
+
+import yaml
+from syft_migration import MigrationService
+
+from syft_job.migrations import job_registry
+from syft_job.models import JobStatus
+
+from .mocks import mock_submission_config_path
+
+FIXTURES_DIR = Path(__file__).parent / "fixtures"
+
+
+def _load_fixture(canonical_name: str, version: str) -> dict:
+    path = FIXTURES_DIR / canonical_name / f"v{version}.yaml"
+    assert path.exists(), (
+        f"Missing fixture {path}; add one for every registered version"
+    )
+    return yaml.safe_load(path.read_text())
+
+
+def test_all_job_state_versions_upgrade_from_disk_to_latest():
+    service = MigrationService(registry=job_registry)
+    latest = job_registry.latest_version("JobState")
+    latest_cls = job_registry.get_class("JobState", latest)
+
+    for version in job_registry.versions("JobState"):
+        upgraded = service.load(
+            _load_fixture("JobState", version), target_version=latest
+        )
+        assert type(upgraded) is latest_cls
+        assert upgraded.status == JobStatus.DONE
+
+
+def test_latest_job_state_downgrades_to_all_versions(tmp_path: Path):
+    service = MigrationService(registry=job_registry)
+    latest = job_registry.latest_version("JobState")
+    newest = service.load(_load_fixture("JobState", latest))
+
+    for version in job_registry.versions("JobState"):
+        version_cls = job_registry.get_class("JobState", version)
+        downgraded = service.migrate(newest, target_version=version)
+        assert type(downgraded) is version_cls
+
+        path = tmp_path / f"state_v{version}.yaml"
+        downgraded.save(path)
+        assert version_cls.load(path) == downgraded
+
+
+def test_all_submission_versions_upgrade_from_disk_to_latest():
+    service = MigrationService(registry=job_registry)
+    latest = job_registry.latest_version("JobSubmissionMetadata")
+    latest_cls = job_registry.get_class("JobSubmissionMetadata", latest)
+
+    for version in job_registry.versions("JobSubmissionMetadata"):
+        upgraded = service.load(
+            _load_fixture("JobSubmissionMetadata", version), target_version=latest
+        )
+        assert type(upgraded) is latest_cls
+        assert upgraded.name == "my.job"
+
+
+def test_latest_submission_downgrades_to_all_versions(tmp_path: Path):
+    service = MigrationService(registry=job_registry)
+    latest = job_registry.latest_version("JobSubmissionMetadata")
+    newest = service.load(_load_fixture("JobSubmissionMetadata", latest))
+
+    for version in job_registry.versions("JobSubmissionMetadata"):
+        version_cls = job_registry.get_class("JobSubmissionMetadata", version)
+        downgraded = service.migrate(newest, target_version=version)
+        assert type(downgraded) is version_cls
+
+        path = mock_submission_config_path(tmp_path / f"v{version}")
+        downgraded.save(path)
+        assert version_cls.load(path) == downgraded