feat: versioned JobManager for reading/writing jobs across protocol versions

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

@@ -4,6 +4,7 @@
 from typing import Optional
 
 from syft_job.config import SyftJobConfig
+from syft_job.manager import JobManager, JobRef
 from syft_job.models import JobState, JobStatus, JobSubmissionMetadata
 
 from syft_bg.common.monitor import Monitor
@@ -29,28 +30,20 @@ def __init__(
         self.job_config = SyftJobConfig.from_syftbox_folder(
             str(self.syftbox_root), do_email
         )
+        self.job_manager = JobManager(config=self.job_config)
 
     def _check_all_entities(self):
         self.process_local_status_changes()
 
     def process_local_status_changes(self):
-        inbox_dir = self.job_config.get_all_submissions_dir(self.do_email)
-        if not inbox_dir.exists():
-            return
-
-        for ds_dir in inbox_dir.iterdir():
-            if not ds_dir.is_dir():
-                continue
-            for job_path in ds_dir.iterdir():
-                if not job_path.is_dir():
-                    continue
-                try:
-                    self._maybe_process_job(job_path)
-                except Exception as e:
-                    print(f"[JobMonitor] Error checking job {job_path.name}: {e}")
-
-    def _maybe_process_job(self, job_path: Path):
-        metadata = self._load_job_metadata(job_path)
+        for ref in self.job_manager.iter_submission_refs(self.do_email):
+            try:
+                self._maybe_process_job(ref)
+            except Exception as e:
+                print(f"[JobMonitor] Error checking job {ref.job_name}: {e}")
+
+    def _maybe_process_job(self, ref: JobRef):
+        metadata = self._load_job_metadata(ref)
         if not metadata:
             return
 
@@ -62,7 +55,7 @@ def _maybe_process_job(self, job_path: Path):
             if success:
                 print(f"[JobMonitor] Sent new job notification: {job_name}")
 
-        review_state = self._load_review_state(ds_email, job_name)
+        review_state = self._load_review_state(ref)
 
         if review_state and review_state.status in (
             JobStatus.APPROVED,
@@ -85,61 +78,40 @@ def _maybe_process_job(self, job_path: Path):
 
     def seed_existing_jobs(self):
         """On fresh state, mark all existing jobs so we don't re-notify old jobs."""
-        inbox_dir = self.job_config.get_all_submissions_dir(self.do_email)
-        if not inbox_dir.exists():
-            return
-
         count = 0
-        for ds_dir in inbox_dir.iterdir():
-            if not ds_dir.is_dir():
+        for ref in self.job_manager.iter_submission_refs(self.do_email):
+            metadata = self._load_job_metadata(ref)
+            if not metadata:
                 continue
-            for job_path in ds_dir.iterdir():
-                if not job_path.is_dir():
-                    continue
-                metadata = self._load_job_metadata(job_path)
-                if not metadata:
-                    continue
-                self.state.mark_notified(metadata.name, "new")
-                review_state = self._load_review_state(
-                    metadata.submitted_by, metadata.name
-                )
-                if review_state:
-                    if review_state.status in (
-                        JobStatus.APPROVED,
-                        JobStatus.RUNNING,
-                        JobStatus.DONE,
-                        JobStatus.FAILED,
-                    ):
-                        self.state.mark_notified(metadata.name, "approved")
-                    if review_state.status == JobStatus.DONE:
-                        self.state.mark_notified(metadata.name, "executed")
-                    if review_state.status == JobStatus.FAILED:
-                        self.state.mark_notified(metadata.name, "failed")
-                count += 1
+            self.state.mark_notified(metadata.name, "new")
+            review_state = self._load_review_state(ref)
+            if review_state:
+                if review_state.status in (
+                    JobStatus.APPROVED,
+                    JobStatus.RUNNING,
+                    JobStatus.DONE,
+                    JobStatus.FAILED,
+                ):
+                    self.state.mark_notified(metadata.name, "approved")
+                if review_state.status == JobStatus.DONE:
+                    self.state.mark_notified(metadata.name, "executed")
+                if review_state.status == JobStatus.FAILED:
+                    self.state.mark_notified(metadata.name, "failed")
+            count += 1
 
         if count:
             print(f"[JobMonitor] Seeded {count} existing jobs on fresh state")
 
-    def _load_review_state(self, ds_email: str, job_name: str) -> Optional[JobState]:
+    def _load_review_state(self, ref: JobRef) -> Optional[JobState]:
         """Load state.yaml from the job's review directory."""
-        review_dir = self.job_config.get_review_job_dir(
-            self.do_email, ds_email, job_name
-        )
-        state_file = review_dir / "state.yaml"
-        if not state_file.exists():
-            return None
         try:
-            return JobState.load(state_file)
+            return self.job_manager.read_state(ref)
         except Exception:
             return None
 
-    def _load_job_metadata(self, job_path: Path) -> Optional[JobSubmissionMetadata]:
-        config_file = job_path / "config.yaml"
-        if not config_file.exists():
-            return None
-
+    def _load_job_metadata(self, ref: JobRef) -> Optional[JobSubmissionMetadata]:
         try:
-            return JobSubmissionMetadata.load(config_file)
+            return self.job_manager.read_submission(ref)
         except Exception as e:
-            print(f"[JobMonitor] Error reading job config {config_file}: {e}")
+            print(f"[JobMonitor] Error reading job config for {ref.job_name}: {e}")
             return None

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

@@ -8,7 +8,8 @@
 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 import JobState, JobStatus, JobSubmissionMetadata
+from syft_job.manager import JobRef
+from syft_job.models import JobState, JobStatus
 
 from syft_enclaves.enclave_job_info import (
     EnclaveJobInfo,
@@ -255,33 +256,19 @@ def receive_jobs(self):
         4. Sets permissions and marks as distributed
         """
         self._manager.job_client.scan_inbox()
-        inbox_dir = self._manager.job_client.config.get_all_submissions_dir(
-            self._manager.email
-        )
-        if not inbox_dir.exists():
-            return
+        job_manager = self._manager.job_client.manager
+        for ref in job_manager.iter_submission_refs(self._manager.email):
+            self._try_distribute_job(ref)
 
-        for ds_dir in inbox_dir.iterdir():
-            if not ds_dir.is_dir():
-                continue
-            for job_dir in ds_dir.iterdir():
-                if not job_dir.is_dir():
-                    continue
-                self._try_distribute_job(ds_dir.name, job_dir)
-
-    def _try_distribute_job(self, ds_email: str, job_dir: Path):
+    def _try_distribute_job(self, ref: JobRef):
         """Distribute a single enclave job to relevant DOs if not yet distributed."""
-        config_path = job_dir / "config.yaml"
-        if not config_path.exists():
-            return
-
-        config = JobSubmissionMetadata.load(config_path)
+        job_manager = self._manager.job_client.manager
+        job_dir = job_manager.submission_dir(ref)
+        config = job_manager.read_submission(ref)
         if config.job_type != "enclave" or not config.datasets:
             return
 
-        review_dir = self._manager.job_client.config.get_review_job_dir(
-            self._manager.email, ds_email, job_dir.name
-        )
+        review_dir = job_manager.review_dir(ref)
         distributed_marker = review_dir / "distributed"
         if distributed_marker.exists():
             return
@@ -296,7 +283,7 @@ def _try_distribute_job(self, ds_email: str, job_dir: Path):
         recipients = list(dict.fromkeys([*submission_dos, *approval_dos]))
         self._forward_job_to_dos(job_dir, recipients)
         self._save_enclave_job_state(review_dir, approval_dos, config.datasets)
-        self._set_job_permissions(job_dir, recipients, approval_dos)
+        self._set_job_permissions(ref, recipients, approval_dos)
         self._forward_approval_files_to_dos(review_dir, approval_dos)
 
         distributed_marker.parent.mkdir(parents=True, exist_ok=True)
@@ -369,22 +356,17 @@ def _save_enclave_job_state(
 
     def _set_job_permissions(
         self,
-        job_dir: Path,
+        ref: JobRef,
         read_dos: list[str],
         approval_dos: list[str],
     ):
         """Grant inbox read to everyone who needs to see the job (referenced +
         approving DOs), and approval-file write to the approving DOs."""
+        job_manager = self._manager.job_client.manager
         datasite = self._manager.syftbox_folder / self._manager.email
         ctx = SyftPermContext(datasite=datasite)
-        inbox_rel = job_dir.relative_to(datasite)
-
-        ds_email = job_dir.parent.name
-        job_name = job_dir.name
-        review_dir = self._manager.job_client.config.get_review_job_dir(
-            self._manager.email, ds_email, job_name
-        )
-        review_rel = review_dir.relative_to(datasite)
+        inbox_rel = job_manager.submission_dir(ref).relative_to(datasite)
+        review_rel = job_manager.review_dir(ref).relative_to(datasite)
 
         for do_email in dict.fromkeys([*read_dos, *approval_dos]):
             ctx.open(inbox_rel).grant_read_access(do_email)

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

@@ -20,6 +20,10 @@ def __init__(self, job_client: JobClient):
     def config(self):
         return self._job_client.config
 
+    @property
+    def manager(self):
+        return self._job_client.manager
+
     @property
     def current_user_email(self):
         return self._job_client.current_user_email

packages/syft-job/README.md

@@ -0,0 +1,29 @@
+# syft-job
+
+Job submission and execution for SyftBox: data scientists submit bash/Python jobs
+into a data owner's inbox, the data owner reviews, approves, and runs them.
+
+## Releasing
+
+On **every** release (after bumping the version), run both release scripts:
+
+```bash
+uv run python scripts/export_release_artifact.py
+uv run python scripts/generate_release_fixture.py
+```
+
+`export_release_artifact.py` always writes
+`src/syft_job/migrations/history/package-artifacts/syft-job-<version>.json`
+(the package's identity + the protocol it speaks), and additionally writes
+`history/protocols/protocol-<n>.json` when the release ships a new protocol
+version. It refuses to run if the job protocol changed without bumping
+`JOB_PROTOCOL_VERSION` (`src/syft_job/migrations/registry.py`).
+
+`generate_release_fixture.py` writes a full SyftBox tree —
+`tests/migrations/p2p/fixtures/syft_job-<version>-protocol<p>_syftbox/` — exactly
+as this release serializes jobs to disk. Commit it: future releases loop over
+these fixtures (`test_older_protocol_compatibility.py`) to prove they can still
+read and round-trip older on-disk data.
+
+Tests check the code against these artifacts: released object versions are
+frozen forever — changing one requires a new version plus migrations.

packages/syft-job/scripts/export_release_artifact.py

@@ -0,0 +1,37 @@
+"""Export the release artifacts for the current syft-job 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_job.migrations.history import PACKAGE_ARTIFACTS_DIR, PROTOCOLS_DIR
+from syft_job.migrations.registry import JOB_PROTOCOL_VERSION, job_registry
+from syft_job.version import __version__
+
+
+def main() -> None:
+    # Import the models so every versioned object is registered.
+    import syft_job  # noqa: F401
+
+    if job_registry.protocol_changed_without_bump():
+        sys.exit(
+            "The job protocol changed compared to the released "
+            f"protocol-{JOB_PROTOCOL_VERSION}.json; bump JOB_PROTOCOL_VERSION "
+            "in syft_job/migrations/registry.py before releasing."
+        )
+
+    info_path = PACKAGE_ARTIFACTS_DIR / f"syft-job-{__version__}.json"
+    job_registry.compute_released_package_protocol_info().save(info_path)
+    print(f"Wrote {info_path}")
+
+    protocol_path = PROTOCOLS_DIR / f"protocol-{JOB_PROTOCOL_VERSION}.json"
+    if not protocol_path.exists():
+        job_registry.compute_released_protocol().save(protocol_path)
+        print(f"Wrote {protocol_path} (new protocol version)")
+
+
+if __name__ == "__main__":
+    main()