fix: syft-bg services run only the job they should

docs/API.md

@@ -255,6 +255,16 @@ reported by name, with the submitter and the reason:
 
 The job stays at `approved`, so it runs on the next call once the versions match.
 
+### `client.run_job(job, share_outputs_with_submitter=False, sync=False)`
+
+Run one approved job. **DO only.** Takes the same options as `process_approved_jobs`.
+Raises if the submitter runs an incompatible version. Pass `sync=True` to sync after.
+
+```python
+job = do_client.jobs["analysis"]
+do_client.run_job(job)
+```
+
 ---
 
 ## Cleanup

packages/syft-bg/README.md

@@ -162,7 +162,7 @@ DO notifications are threaded per job (new → approved/rejected → completed i
 Runs api calls, and approves peers based on your config:
 
 - **Peers**: Auto-accept connection requests from approved domains, and from the peers of your apis
-- **Jobs**: Run every approved job; an api call arrives approved
+- **Jobs**: Run api calls, which arrive approved. Jobs you approve by hand wait for you to run them
 
 ## Configuration
 

packages/syft-bg/src/syft_bg/approve/handlers/job.py

@@ -5,6 +5,8 @@
 from pathlib import Path
 from typing import TYPE_CHECKING, Callable, Optional, Protocol
 
+from syft.sync.version import VersionError
+from syft_job.api_requests import API_APPROVAL_METHOD
 from syft_job.job import JobInfo
 
 from syft_bg.approve.config import AutoApprovalsConfig
@@ -22,10 +24,10 @@ def was_approved(self, job_name: str) -> bool: ...
 
 
 class ApprovedJobHandler:
-    """Runs the approved jobs on the DO's datasite.
+    """Runs the api calls on the DO's datasite.
 
     An api call arrives approved: receiving it turns the request into an
-    approved job. A job the DO approved by hand runs too.
+    approved job. A job the DO approved by hand waits for the DO to run it.
     """
 
     def __init__(
@@ -48,27 +50,36 @@ def config(self) -> AutoApprovalsConfig:
         return SyftBgConfig.load(self._config_path).approve.auto_approvals
 
     def run_approved_jobs(self) -> list[JobInfo]:
-        """Receive new jobs, then run every approved one.
+        """Receive new jobs, then run every approved api call.
 
-        Returns the approved jobs not seen before.
+        Returns the approved api calls not seen before.
         """
         if not self.config.enabled:
             return []
         self.client.receive_jobs()
-        approved = [
+        api_calls = [
             job
             for job in self.client.job_client.jobs
             if job.status == "approved"
             and job.datasite_owner_email == self.client.email
+            and job.approval_method == API_APPROVAL_METHOD
         ]
-        new = self._record_new(approved)
-        if approved:
-            self.client.process_approved_jobs(
-                stream_output=self.verbose,
-                share_outputs_with_submitter=True,
-            )
+        new = self._record_new(api_calls)
+        for job in api_calls:
+            self._run(job)
+        if api_calls:
+            self.client.sync()
         return new
 
+    def _run(self, job: JobInfo) -> None:
+        """Run one api call; a submitter on an incompatible version is skipped."""
+        try:
+            self.client.run_job(
+                job, stream_output=self.verbose, share_outputs_with_submitter=True
+            )
+        except VersionError as e:
+            print(f"Skipped: {job.name} from {job.submitted_by}: {e}")
+
     def _record_new(self, approved: list[JobInfo]) -> list[JobInfo]:
         """Mark the jobs not seen before, and call on_approve for each."""
         new = [

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

@@ -14,7 +14,7 @@
 
 
 class JobMonitor(Monitor):
-    """Watches for approved jobs, and runs them."""
+    """Watches for approved api calls, and runs them."""
 
     def __init__(
         self,
@@ -34,5 +34,5 @@ def __init__(
         )
 
     def _check_all_entities(self):
-        """Run approved jobs."""
+        """Run approved api calls."""
         self.handler.run_approved_jobs()

packages/syft-bg/src/syft_bg/email_approve/handler.py

@@ -127,7 +127,9 @@ def _approve_job(
         )
         self.state.mark_notified(state_key, "processed")
 
-        self.job_runner.process_approved_jobs(share_outputs_with_submitter=True)
+        self.job_runner.run_job(
+            job.name, job.submitted_by, share_outputs_with_submitter=True
+        )
         print(f"[EmailApproveHandler] Approved job: {job_name}")
 
     def _approve_and_create_api(self, job, job_name: str, state_key: str) -> None:

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

@@ -906,19 +906,69 @@ def process_approved_jobs(
         print(f" Found {len(approved_jobs)} job(s) in approved status")
 
         for ref in approved_jobs:
-            print(f"\n{'=' * 50}")
-            if self._execute_job(ref, stream_output=stream_output, timeout=timeout):
-                self._share_job_results(
-                    ref,
-                    share_outputs_with_submitter,
-                    share_logs_with_submitter,
-                    self._output_recipients(ref, approved_peers),
-                )
-            print(f"{'=' * 50}")
+            self._run_job(
+                ref,
+                stream_output,
+                timeout,
+                share_outputs_with_submitter,
+                share_logs_with_submitter,
+                approved_peers,
+            )
 
         if approved_jobs:
             print(f"\n Processed {len(approved_jobs)} job(s)")
 
+    def run_job(
+        self,
+        job_name: str,
+        ds_email: str,
+        stream_output: bool = True,
+        timeout: int | None = None,
+        share_outputs_with_submitter: bool = False,
+        share_logs_with_submitter: bool = False,
+        approved_peers: list[str] | None = None,
+    ) -> None:
+        """Run one approved job, and share its results as asked.
+
+        Takes the same options as ``process_approved_jobs``. Raises if the job
+        is not approved.
+        """
+        ref = self._find_jobref_from_name(job_name, ds_email)
+        status = self._get_job_state(ref).status
+        if status != JobStatus.APPROVED:
+            raise ValueError(
+                f"Job '{job_name}' from {ds_email} is {status.value}, not approved"
+            )
+        self._replace_ds_folder_reads()
+        self._run_job(
+            ref,
+            stream_output,
+            timeout,
+            share_outputs_with_submitter,
+            share_logs_with_submitter,
+            approved_peers,
+        )
+
+    def _run_job(
+        self,
+        ref: JobRef,
+        stream_output: bool,
+        timeout: int | None,
+        share_outputs: bool,
+        share_logs: bool,
+        approved_peers: list[str] | None,
+    ) -> None:
+        """Execute an approved job, then share its results if it ran."""
+        print(f"\n{'=' * 50}")
+        if self._execute_job(ref, stream_output=stream_output, timeout=timeout):
+            self._share_job_results(
+                ref,
+                share_outputs,
+                share_logs,
+                self._output_recipients(ref, approved_peers),
+            )
+        print(f"{'=' * 50}")
+
     def share_job_results(
         self,
         job_name: str,

packages/syft-job/tests/test_job_flow.py

@@ -317,6 +317,38 @@ def test_skipping_one_job_spares_its_same_named_sibling(tmp_path: Path):
     assert final[DS2_EMAIL] == "done", "its same-named sibling must still run"
 
 
+def test_run_job_runs_only_that_job(tmp_path: Path):
+    """run_job runs the named job; another approved job stays approved."""
+    syftbox = tmp_path / "SyftBox"
+    syftbox.mkdir()
+    code_file = tmp_path / "main.py"
+    code_file.write_text(MAIN_PY)
+
+    do_config = SyftJobConfig(syftbox_folder=syftbox, current_user_email=DO_EMAIL)
+    do_client = JobClient(config=do_config)
+    do_runner = SyftJobRunner(config=do_config)
+    ds_client = JobClient(
+        config=SyftJobConfig(syftbox_folder=syftbox, current_user_email=DS_EMAIL)
+    )
+    for name in ("run.job", "wait.job", "pending.job"):
+        ds_client.submit_python_job(
+            user=DO_EMAIL, code_path=str(code_file), job_name=name
+        )
+    do_client.jobs["run.job"].approve()
+    do_client.jobs["wait.job"].approve()
+
+    do_runner.run_job("run.job", DS_EMAIL, stream_output=False, timeout=60)
+
+    final = {job.name: job.status for job in do_client.jobs}
+    assert final == {
+        "run.job": "done",
+        "wait.job": "approved",
+        "pending.job": "pending",
+    }
+    with pytest.raises(ValueError, match="is pending, not approved"):
+        do_runner.run_job("pending.job", DS_EMAIL)
+
+
 def test_runner_ignores_approved_jobs_on_another_datasite(tmp_path: Path):
     """`jobs` spans every datasite in the folder; the runner runs only its own.
 

packages/syft-rds/src/syft_rds/client.py

@@ -24,6 +24,7 @@
 from syft.sync.utils.pre_submit_scan import run_pre_submit_check
 from syft.sync.version.peer_manager import CompatAction
 from syft_job.client import JobClient
+from syft_job.job import JobInfo
 from syft_job.job_runner import SyftJobRunner
 from syft_datasets.dataset_manager import SHARE_WITH_ANY, SyftDatasetManager
 from syft_datasets.dataset_ref import DatasetNotFoundError
@@ -669,11 +670,7 @@ def process_approved_jobs(
         ignore_peer_version: bool = False,
     ) -> None:
         """Process approved jobs (DO only). Auto-syncs after unless PRE_SYNC=false."""
-        if not self.has_do_role:
-            raise ValueError("Only dataset owners can process approved jobs")
-        if self.job_runner is None:
-            raise ValueError("Job runner is not configured for this client")
-
+        job_runner = self._job_runner_for_do()
         skipped: list[tuple[str, str, str]] = []
 
         if not force_execution:
@@ -703,13 +700,13 @@ def process_approved_jobs(
                         )
                     )
 
-        self.job_runner.process_approved_jobs(
+        job_runner.process_approved_jobs(
             stream_output=stream_output,
             timeout=timeout,
             skip_jobs=[(name, peer) for name, peer, _ in skipped] or None,
             share_outputs_with_submitter=share_outputs_with_submitter,
             share_logs_with_submitter=share_logs_with_submitter,
-            approved_peers=[p.email for p in self.peer_manager.approved_peers],
+            approved_peers=self._approved_peer_emails(),
         )
 
         # maybe_warn() above logs the peer, never the job.
@@ -719,6 +716,52 @@ def process_approved_jobs(
         if self._pre_sync_enabled:
             self.sync_engine.sync()
 
+    def run_job(
+        self,
+        job: JobInfo,
+        stream_output: bool = True,
+        timeout: int | None = None,
+        share_outputs_with_submitter: bool = False,
+        share_logs_with_submitter: bool = False,
+        ignore_peer_version: bool = False,
+        sync: bool = False,
+    ) -> None:
+        """Run one approved job (DO only).
+
+        Raises if the submitter's version is incompatible. Syncs after only
+        with ``sync=True``, so a caller running several jobs syncs once.
+        """
+        job_runner = self._job_runner_for_do()
+        result = self.sync_engine.peer_manager.get_peer_compatibility_status(
+            job.submitted_by,
+            action=CompatAction.EXECUTE,
+            ignore_peer_version=ignore_peer_version,
+        )
+        result.raise_on_skip(operation="run job")
+        result.maybe_warn()
+        job_runner.run_job(
+            job.name,
+            job.submitted_by,
+            stream_output=stream_output,
+            timeout=timeout,
+            share_outputs_with_submitter=share_outputs_with_submitter,
+            share_logs_with_submitter=share_logs_with_submitter,
+            approved_peers=self._approved_peer_emails(),
+        )
+        if sync:
+            self.sync_engine.sync()
+
+    def _job_runner_for_do(self) -> SyftJobRunner:
+        """The job runner; raises unless this is a DO client that has one."""
+        if not self.has_do_role:
+            raise ValueError("Only dataset owners can run jobs")
+        if self.job_runner is None:
+            raise ValueError("Job runner is not configured for this client")
+        return self.job_runner
+
+    def _approved_peer_emails(self) -> list[str]:
+        return [p.email for p in self.peer_manager.approved_peers]
+
     # ------------------------------------------------------------------ #
     # dataset product surface (RDS-owned)
     # ------------------------------------------------------------------ #

packages/syft-rds/tests/test_version_negotiation.py

@@ -4,7 +4,7 @@
 
 import pytest
 
-from syft.sync.version.exceptions import VersionUnknownError
+from syft.sync.version.exceptions import VersionMismatchError, VersionUnknownError
 from syft.sync.encryption_policy import EncryptionPolicy
 from syft.sync.version.peer_info import PeerInfo
 from syft.sync.version.peer_manager import PeerCompatibilityResult, PeerManager
@@ -176,3 +176,31 @@ def test_skipped_job_name_and_reason_reach_stdout(self, tmp_path, capfd):
         assert "incompatible version" in out
         assert "ignore_peer_version=True" in out
         assert do.jobs["skipped.job"].status == "approved"
+
+
+def test_run_job_raises_on_incompatible_submitter(tmp_path):
+    """A single run raises instead of skipping; the job stays approved."""
+    ds, do = SyftRDSClient.pair_with_mock_drive_service_connection(
+        use_in_memory_cache=False,
+        sync_automatically=False,
+    )
+    code_path = tmp_path / "blocked.py"
+    code_path.write_text('print("hello")')
+    ds.submit_python_job(
+        user=do.email, code_path=str(code_path), job_name="blocked.job"
+    )
+    do.sync()
+    do.jobs["blocked.job"].approve()
+
+    incompatible = PeerCompatibilityResult(
+        peer_email=ds.email,
+        status=CompatibilityStatus.INCOMPATIBLE,
+        should_skip=True,
+        explanation_skip="incompatible version",
+    )
+    with patch.object(
+        PeerManager, "get_peer_compatibility_status", return_value=incompatible
+    ):
+        with pytest.raises(VersionMismatchError):
+            do.run_job(do.jobs["blocked.job"])
+    assert do.jobs["blocked.job"].status == "approved"