From 5720490023be05f3299a056f2ef94e259078f0af Mon Sep 17 00:00:00 2001 From: Koen van der Veen Date: Wed, 4 Nov 2026 15:23:49 +0100 Subject: [PATCH] fix: syft-bg services run only the job they should The approve service runs only api calls; jobs approved by hand wait for the data owner. An email approval runs and shares only the job replied to. Adds run_job to SyftJobRunner and SyftRDSClient to run a single approved job. --- docs/API.md | 10 +++ packages/syft-bg/README.md | 2 +- .../src/syft_bg/approve/handlers/job.py | 33 ++++++--- .../src/syft_bg/approve/monitors/job.py | 4 +- .../src/syft_bg/email_approve/handler.py | 4 +- packages/syft-job/src/syft_job/job_runner.py | 68 ++++++++++++++++--- packages/syft-job/tests/test_job_flow.py | 32 +++++++++ packages/syft-rds/src/syft_rds/client.py | 57 ++++++++++++++-- .../tests/test_version_negotiation.py | 30 +++++++- tests/unit/syft_bg/test_create_api.py | 42 +++++++++--- .../unit/syft_bg/test_email_approval_flow.py | 43 ++++++++++++ tests/unit/syft_bg/test_email_approve.py | 12 ++-- 12 files changed, 290 insertions(+), 47 deletions(-) diff --git a/docs/API.md b/docs/API.md index 6ef905bd580..9dfa916b12e 100644 --- a/docs/API.md +++ b/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 diff --git a/packages/syft-bg/README.md b/packages/syft-bg/README.md index 11ab1e30946..cecef816820 100644 --- a/packages/syft-bg/README.md +++ b/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 diff --git a/packages/syft-bg/src/syft_bg/approve/handlers/job.py b/packages/syft-bg/src/syft_bg/approve/handlers/job.py index 87388a9b00a..17b94dc8c7f 100644 --- a/packages/syft-bg/src/syft_bg/approve/handlers/job.py +++ b/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 = [ diff --git a/packages/syft-bg/src/syft_bg/approve/monitors/job.py b/packages/syft-bg/src/syft_bg/approve/monitors/job.py index 81b35fe5e08..f2fde554565 100644 --- a/packages/syft-bg/src/syft_bg/approve/monitors/job.py +++ b/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() diff --git a/packages/syft-bg/src/syft_bg/email_approve/handler.py b/packages/syft-bg/src/syft_bg/email_approve/handler.py index 7abd96cdbb6..46f1a7a04fd 100644 --- a/packages/syft-bg/src/syft_bg/email_approve/handler.py +++ b/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: diff --git a/packages/syft-job/src/syft_job/job_runner.py b/packages/syft-job/src/syft_job/job_runner.py index 9668ed3b357..93da88fea37 100644 --- a/packages/syft-job/src/syft_job/job_runner.py +++ b/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, diff --git a/packages/syft-job/tests/test_job_flow.py b/packages/syft-job/tests/test_job_flow.py index 019f27db871..07dc4d87e23 100644 --- a/packages/syft-job/tests/test_job_flow.py +++ b/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. diff --git a/packages/syft-rds/src/syft_rds/client.py b/packages/syft-rds/src/syft_rds/client.py index 0d0fe2eb1ae..f3da69b2bc8 100644 --- a/packages/syft-rds/src/syft_rds/client.py +++ b/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) # ------------------------------------------------------------------ # diff --git a/packages/syft-rds/tests/test_version_negotiation.py b/packages/syft-rds/tests/test_version_negotiation.py index c942ffe0e35..d35a4571228 100644 --- a/packages/syft-rds/tests/test_version_negotiation.py +++ b/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" diff --git a/tests/unit/syft_bg/test_create_api.py b/tests/unit/syft_bg/test_create_api.py index 2195b9cbecd..32fffa15b8f 100644 --- a/tests/unit/syft_bg/test_create_api.py +++ b/tests/unit/syft_bg/test_create_api.py @@ -24,6 +24,7 @@ from syft_bg.approve.handlers.job import ApprovedJobHandler from syft_bg.common.config import get_default_paths from syft_bg.common.syft_bg_config import SyftBgConfig +from syft.sync.version import VersionUnknownError DO_EMAIL = "do@test.com" PINNED = ["run.sh", "code/main.py"] @@ -294,8 +295,18 @@ def test_argument_overrides_config(self, temp_dir): assert store.get("main").disclosures == ["logs"] +def _approved_job( + approval_method: str = "api", datasite_owner_email: str = DO_EMAIL +) -> MagicMock: + return MagicMock( + status="approved", + datasite_owner_email=datasite_owner_email, + approval_method=approval_method, + ) + + class TestApprovedJobHandler: - """The approve service runs what is approved; calls arrive approved.""" + """The approve service runs api calls; they arrive approved.""" def _handler( self, temp_dir: Path, jobs: list @@ -307,24 +318,35 @@ def _handler( ) return handler, client - def test_runs_approved_jobs_on_own_datasite(self, temp_dir): - call = MagicMock(status="approved", datasite_owner_email=DO_EMAIL) + def test_runs_api_calls_on_own_datasite(self, temp_dir): + call = _approved_job() handler, client = self._handler(temp_dir, [call]) assert handler.run_approved_jobs() == [call] client.receive_jobs.assert_called_once() - client.process_approved_jobs.assert_called_once() - assert "share_logs_with_submitter" not in ( - client.process_approved_jobs.call_args.kwargs + client.run_job.assert_called_once_with( + call, stream_output=False, share_outputs_with_submitter=True ) + client.sync.assert_called_once() - def test_runs_nothing_without_approved_jobs(self, temp_dir): + def test_runs_nothing_without_approved_api_calls(self, temp_dir): pending = MagicMock(status="pending", datasite_owner_email=DO_EMAIL) - elsewhere = MagicMock(status="approved", datasite_owner_email="other@test.com") - handler, client = self._handler(temp_dir, [pending, elsewhere]) + elsewhere = _approved_job(datasite_owner_email="other@test.com") + by_hand = _approved_job(approval_method="manual") + handler, client = self._handler(temp_dir, [pending, elsewhere, by_hand]) assert handler.run_approved_jobs() == [] - client.process_approved_jobs.assert_not_called() + client.run_job.assert_not_called() + client.sync.assert_not_called() + + def test_incompatible_submitter_does_not_stop_next_call(self, temp_dir): + blocked, call = _approved_job(), _approved_job() + handler, client = self._handler(temp_dir, [blocked, call]) + client.run_job.side_effect = [VersionUnknownError("ds@test.com", "run"), None] + + handler.run_approved_jobs() + + assert [c.args[0] for c in client.run_job.call_args_list] == [blocked, call] class TestJobUserFiles: diff --git a/tests/unit/syft_bg/test_email_approval_flow.py b/tests/unit/syft_bg/test_email_approval_flow.py index 7fecde67bc8..0fcd3880e75 100644 --- a/tests/unit/syft_bg/test_email_approval_flow.py +++ b/tests/unit/syft_bg/test_email_approval_flow.py @@ -6,6 +6,7 @@ from pathlib import Path from unittest.mock import MagicMock +from syft_bg.approve.handlers.job import ApprovedJobHandler from syft_bg.common.state import JsonStateManager from syft_bg.email_approve.config import EmailApproveConfig from syft_bg.email_approve.gmail_message import GmailMessage @@ -221,3 +222,45 @@ def test_email_approval_e2e(): assert email_approve_state.was_notified( f"email_reply_{FAKE_THREAD_ID}", "processed" ) + + +def test_services_leave_hand_approved_job_alone(tmp_path): + """An email reply runs only its job; the approve service runs no hand-approved job.""" + ds_manager, do_manager = SyftRDSClient.pair_with_mock_drive_service_connection( + use_in_memory_cache=False, + sync_automatically=False, + ) + code_path = tmp_path / "main.py" + code_path.write_text( + "import os\n" + "os.makedirs('outputs', exist_ok=True)\n" + "open('outputs/result.txt', 'w').write('done')\n" + ) + for name in ("by_hand.job", "emailed.job"): + ds_manager.submit_python_job( + user=do_manager.email, code_path=str(code_path), job_name=name + ) + do_manager.sync() + do_manager.jobs["by_hand.job"].approve() + + notify_state = JsonStateManager(state_file=tmp_path / "notify_state.json") + notify_state.store_thread_id("emailed.job", FAKE_THREAD_ID) + EmailApproveHandler( + job_client=do_manager.job_client, + job_runner=do_manager.job_runner, + state=JsonStateManager(state_file=tmp_path / "email_approve_state.json"), + notify_state=notify_state, + do_email=do_manager.email, + ).handle_reply(thread_id=FAKE_THREAD_ID, reply_text="approve") + + statuses = {job.name: job.status for job in do_manager.jobs} + assert statuses == {"by_hand.job": "approved", "emailed.job": "done"} + do_manager.sync() + ds_manager.sync() + assert ds_manager.jobs["emailed.job"].output_paths + + approve_service = ApprovedJobHandler( + client=do_manager, config_path=tmp_path / "config.yaml", verbose=False + ) + assert approve_service.run_approved_jobs() == [] + assert do_manager.jobs["by_hand.job"].status == "approved" diff --git a/tests/unit/syft_bg/test_email_approve.py b/tests/unit/syft_bg/test_email_approve.py index a0b426be8b5..4a0287cc1f5 100644 --- a/tests/unit/syft_bg/test_email_approve.py +++ b/tests/unit/syft_bg/test_email_approve.py @@ -156,9 +156,8 @@ def test_approve_job(self, tmp_path, kwargs, disclosures): notify_state.store_thread_id("test.job", "thread123") - mock_job = MagicMock() + mock_job = MagicMock(status="pending", submitted_by="ds@example.com") mock_job.name = "test.job" - mock_job.status = "pending" job_client.jobs = [mock_job] handler.handle_reply(thread_id="thread123", reply_text="approve") @@ -166,9 +165,10 @@ def test_approve_job(self, tmp_path, kwargs, disclosures): mock_job.approve.assert_called_once_with( approval_method="manual", disclosures=disclosures ) - job_runner.process_approved_jobs.assert_called_once_with( - share_outputs_with_submitter=True + job_runner.run_job.assert_called_once_with( + "test.job", "ds@example.com", share_outputs_with_submitter=True ) + job_runner.process_approved_jobs.assert_not_called() def test_deny_job(self, tmp_path): handler, job_client, _, state, notify_state = self._make_handler(tmp_path) @@ -246,7 +246,9 @@ def test_auto_approve_reply_approves_and_creates_api(self, tmp_path): handler.handle_reply(thread_id="thread_auto", reply_text="auto-approve") mock_job.approve.assert_called_once() - job_runner.process_approved_jobs.assert_called_once() + job_runner.run_job.assert_called_once_with( + "test.job", "ds@example.com", share_outputs_with_submitter=True + ) mock_create_api.assert_called_once_with(mock_job) def test_job_not_pending(self, tmp_path):