From 41f029e9f76945a4d82ef6214f1ab6c948b55337 Mon Sep 17 00:00:00 2001 From: Pedro Werneck Date: Fri, 30 Oct 2026 11:53:38 -0300 Subject: [PATCH 1/6] feat: stage job outputs until a release The runner moves the outputs to staging/, like the logs. A reader gets them only through JobInfo.release(filenames, readers), which replaces _release_to_submitter, share_outputs and share_logs. release also grants the files that an earlier release moved, so a later share reaches the DS after an output policy released the outputs to other readers. The enclave releases the outputs before it forwards them, and writes the receipt into the staged outputs. --- .../syft-enclave/src/syft_enclaves/client.py | 3 +- .../src/syft_enclaves/receipt/writer.py | 6 +- packages/syft-job/src/syft_job/job.py | 60 +++++++++---------- packages/syft-job/src/syft_job/job_runner.py | 53 ++++++---------- packages/syft-job/tests/test_job_flow.py | 10 ++-- packages/syft-job/tests/test_staging.py | 2 +- tests/unit/test_sync_manager.py | 2 +- 7 files changed, 55 insertions(+), 81 deletions(-) diff --git a/packages/syft-enclave/src/syft_enclaves/client.py b/packages/syft-enclave/src/syft_enclaves/client.py index 10cd91b08ab..d5fc786a025 100644 --- a/packages/syft-enclave/src/syft_enclaves/client.py +++ b/packages/syft-enclave/src/syft_enclaves/client.py @@ -506,13 +506,14 @@ def _share_results_once(self, job: JobInfo) -> None: self._try_write_receipt(job) # Always share results with the DS (submitter) + job.release(["outputs"], [job.submitted_by]) self._forward_results_to_recipients(job, [job.submitted_by]) # Optionally share with DOs datasets = job.job_metadata.datasets if job.job_headers.get("share_results_with_do") and datasets: do_emails = list(datasets.keys()) - job.share_outputs(do_emails) + job.release(["outputs"], do_emails) self._forward_results_to_recipients(job, do_emails) marker.write_text("shared") diff --git a/packages/syft-enclave/src/syft_enclaves/receipt/writer.py b/packages/syft-enclave/src/syft_enclaves/receipt/writer.py index efa4149c886..624a3fbce28 100644 --- a/packages/syft-enclave/src/syft_enclaves/receipt/writer.py +++ b/packages/syft-enclave/src/syft_enclaves/receipt/writer.py @@ -71,7 +71,7 @@ def write_receipt( public_key = signing_key.public_key().public_bytes_raw() receipt = _receipt(client, job, settings, public_key) envelope = sign_receipt(receipt, signing_key) - path = job.job_review_path / "outputs" / RECEIPT_FILE_NAME + path = job.artifact_path("outputs") / RECEIPT_FILE_NAME path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(envelope, indent=2)) logger.info("Wrote signed receipt for job %s to %s", job.name, path) @@ -80,7 +80,7 @@ def write_receipt( def write_receipt_error(job: JobInfo) -> None: """Record the current exception in the job's outputs.""" - path = job.job_review_path / "outputs" / RECEIPT_ERROR_FILE_NAME + path = job.artifact_path("outputs") / RECEIPT_ERROR_FILE_NAME path.parent.mkdir(parents=True, exist_ok=True) path.write_text(traceback.format_exc()) @@ -91,7 +91,7 @@ def _receipt( settings: ReceiptSettings, run_public_key: bytes, ) -> dict[str, Any]: - outputs_dir = job.job_review_path / "outputs" + outputs_dir = job.artifact_path("outputs") job_entry = _job(job) owners = job.job_metadata.datasets or {} shared = bool(job.job_headers.get("share_results_with_do")) diff --git a/packages/syft-job/src/syft_job/job.py b/packages/syft-job/src/syft_job/job.py index 2c3e2fa1672..7533d9de36e 100644 --- a/packages/syft-job/src/syft_job/job.py +++ b/packages/syft-job/src/syft_job/job.py @@ -97,28 +97,31 @@ def artifact_path(self, filename: str) -> Path: return self.job_staging_path / filename def release_logs(self) -> list[str]: - """Release the logs and the exit code to the submitter. + """Move the logs and the exit code to review/, for the submitter to read. - Returns the names moved. A copy that already reached another party does - not come back. + Returns the names that moved. This call cannot take back a copy that + another party already has. """ - return self._release_to_submitter(STAGED_LOG_FILES) + return self.release(STAGED_LOG_FILES, [self.submitted_by]) - def _release_to_submitter(self, filenames: Iterable[str]) -> list[str]: - """Move the named staged files into review/, and grant the submitter read. + def release(self, filenames: Iterable[str], readers: list[str]) -> list[str]: + """Move the named staged files to review/, and give ``readers`` read access. - The move discloses a file to a reader with the review folder grant. The - file grant covers a reader that holds no folder grant. + Returns the names that moved. A file that is already in review/ also + gets the read access, so a caller can add a reader later. A reader with + read access on the whole review folder also sees each file that moves. """ + filenames = list(filenames) moved = self.release_artifacts(filenames) - self._grant_read(moved, [self.submitted_by]) + released = [n for n in filenames if (self.job_review_path / n).exists()] + self._grant_read(released, readers) return moved def release_artifacts(self, filenames: Iterable[str]) -> list[str]: - """Move the named staged artifacts into the review directory. + """Move the named staged files to review/, and give no read access. - Returns the names it moved. A name that is absent from staging, or that - a release already moved, is skipped. + Returns the names that moved. The method skips a name that is not in + staging/, for example a file that moved before. """ moved = [] for filename in filenames: @@ -205,7 +208,7 @@ def update_disclosures(self, disclosures: DisclosuresArg) -> dict[str, bool]: def release_disclosures(self) -> list[str]: """Release the granted items to the submitter, and return the names moved.""" - return self._release_to_submitter(gated_names(self.granted_disclosures)) + return self.release(gated_names(self.granted_disclosures), [self.submitted_by]) # ────────────────────────────────────────────── # Properties from config (inbox/) @@ -265,7 +268,7 @@ def output_paths(self) -> List[Path]: status = self._state.status def _list_output_files() -> List[Path]: - outputs_dir = self.job_review_path / "outputs" + outputs_dir = self.artifact_path("outputs") if not outputs_dir.exists(): return [] try: @@ -533,7 +536,7 @@ def accept_by_depositing_result(self, path: str) -> Path: path: Path to the result file or folder to deposit Returns: - Path to the deposited result in the review/outputs directory + Path to the deposited result in the staging/outputs directory Raises: ValueError: If job is not in pending or approved status @@ -548,8 +551,8 @@ def accept_by_depositing_result(self, path: str) -> Path: if not result_path.exists(): raise FileNotFoundError(f"Result path not found: {path}") - # Create outputs directory in review/ - outputs_dir = self.job_review_path / "outputs" + # The result waits in staging/ until a release, like a run's outputs. + outputs_dir = self.job_staging_path / "outputs" outputs_dir.mkdir(parents=True, exist_ok=True) # Handle both files and folders @@ -608,10 +611,13 @@ def rerun(self) -> None: f.unlink() changes_made.append(filename) - outputs_dir = self.job_review_path / "outputs" - if outputs_dir.exists() and outputs_dir.is_dir(): - shutil.rmtree(outputs_dir) - changes_made.append("outputs directory") + for outputs_dir in ( + self.job_review_path / "outputs", + self.job_staging_path / "outputs", + ): + if outputs_dir.is_dir(): + shutil.rmtree(outputs_dir) + changes_made.append("outputs directory") # Reset state to approved self._state.status = JobStatus.APPROVED @@ -643,18 +649,6 @@ def _relative_review_path(self, subpath: str) -> Path: ) return rel / subpath - def share_outputs(self, users: list[str]) -> None: - """Grant read access to the outputs directory for given users.""" - ctx = self._get_perm_context() - outputs_rel = self._relative_review_path("outputs") - folder = ctx.open(outputs_rel) - for user in users: - folder.grant_read_access(user) - - def share_logs(self, users: list[str]) -> None: - """Grant read access to log files (stdout, stderr, returncode) for given users.""" - self._grant_read(STAGED_LOG_FILES, users) - def _grant_read(self, filenames: Iterable[str], users: list[str]) -> None: """Grant read access to the named review/ files for the given users.""" ctx = self._get_perm_context() diff --git a/packages/syft-job/src/syft_job/job_runner.py b/packages/syft-job/src/syft_job/job_runner.py index 9668ed3b357..9abfa90579d 100644 --- a/packages/syft-job/src/syft_job/job_runner.py +++ b/packages/syft-job/src/syft_job/job_runner.py @@ -597,7 +597,6 @@ def _run_approved_job( policies: JobPolicies | None = None, ) -> None: job_name = ref.job_name - review_dir = self.manager.review_dir(ref) self._prepare_outputs_dir(ref, run_dir) print(f" Executing job: {job_name}") @@ -615,8 +614,7 @@ def _run_approved_job( self._capture_traceback(ref, returncode, run_dir / "code") returncode = self._run_output_policy(ref, run_dir, policies, returncode) - # Move outputs from the run copy to review/ - self._move_outputs_to_review(run_dir, review_dir) + self._move_outputs_to_staging(ref, run_dir) self._finalize(ref, returncode) self._report_result(ref, returncode) @@ -652,7 +650,7 @@ def _run_output_policy( """Run the output policy, record who gets the outputs, return the exit code. A failed job or a deny shares the outputs with nobody; a deny also - fails the job. The outputs still go to review/, for the data owner. + fails the job. The outputs still go to staging/, for the data owner. """ if policies is None or policies.output is None: return returncode @@ -688,7 +686,7 @@ def _run_policy( def _deny(self, ref: JobRef, kind: PolicyKind, result: PolicyResult) -> None: """Release the reason of a deny to the submitter.""" self._write_staged(ref, POLICY_ERROR_FILENAME, f"{result.error}\n") - self._get_job_info(ref)._release_to_submitter([POLICY_ERROR_FILENAME]) + self._get_job_info(ref).release([POLICY_ERROR_FILENAME], [ref.ds_email]) print(f" Job {ref.job_name} denied by the {kind.value} policy: {result.error}") def _write_recipients(self, ref: JobRef, recipients: list[str]) -> None: @@ -742,23 +740,14 @@ def _finalize(self, ref: JobRef, returncode: int) -> None: state.return_code = None self.manager.write_state(ref, state) - def _move_outputs_to_review(self, run_dir: Path, review_dir: Path) -> None: - run_outputs = run_dir / "code" / "outputs" - review_outputs = review_dir / "outputs" - if run_outputs.exists() and run_outputs.is_dir(): - # Merge into review/outputs (which was pre-created by _prepare_outputs_dir) - for item in run_outputs.iterdir(): - dest = review_outputs / item.name - if item.is_file(): - shutil.copy2(str(item), str(dest)) - elif item.is_dir(): - if dest.exists(): - shutil.rmtree(dest) - shutil.copytree(str(item), str(dest)) - shutil.rmtree(run_outputs) + def _move_outputs_to_staging(self, ref: JobRef, run_dir: Path) -> None: + """Move the outputs from the run copy to staging/, to wait for a release.""" + staging_dir = self.manager.staging_dir(ref) + staging_dir.mkdir(parents=True, exist_ok=True) + shutil.move(str(run_dir / "code" / "outputs"), str(staging_dir / "outputs")) def _prepare_outputs_dir(self, ref: JobRef, run_dir: Path) -> None: - """Create outputs/ in the run copy (for the job) and review/ (for results). + """Create outputs/ in the run copy, and remove the outputs of a previous run. The job writes its outputs in the run copy, outside the SyftBox, so no sync can send them to the submitter before they are released. @@ -766,20 +755,12 @@ def _prepare_outputs_dir(self, ref: JobRef, run_dir: Path) -> None: # Create outputs/ inside code/ dir so job scripts can write there (cwd is code/) (run_dir / "code" / "outputs").mkdir(parents=True, exist_ok=True) - # Create outputs/ in review dir with owner-only read permissions - review_dir = self.manager.review_dir(ref) - outputs_dir = review_dir / "outputs" - if outputs_dir.exists(): - shutil.rmtree(outputs_dir) - outputs_dir.mkdir(parents=True, exist_ok=True) - - from syft_perms import SyftPermContext - - datasite = self.config.syftbox_folder / self.config.current_user_email - rel_path = str(outputs_dir.relative_to(datasite)) + "/" - ctx = SyftPermContext(datasite=datasite) - folder = ctx.open(rel_path) - folder.grant_read_access(self.config.current_user_email) + for outputs_dir in ( + self.manager.review_dir(ref) / "outputs", + self.manager.staging_dir(ref) / "outputs", + ): + if outputs_dir.exists(): + shutil.rmtree(outputs_dir) def _replace_ds_folder_reads(self) -> None: """Replace the read grant on a whole DS review folder with the named files. @@ -946,13 +927,13 @@ def _share_job_results( job_info = self._get_job_info(ref) job_info.release_disclosures() if share_outputs: - job_info.share_outputs(recipients) + job_info.release(["outputs"], recipients) if ref.ds_email not in recipients: # An empty marker: that the submitter can read this file is # the message. Their job then shows "Outputs not shared with # you" instead of looking like it wrote no outputs. self._write_staged(ref, OUTPUTS_WITHHELD_FILENAME, "") - job_info._release_to_submitter([OUTPUTS_WITHHELD_FILENAME]) + job_info.release([OUTPUTS_WITHHELD_FILENAME], [ref.ds_email]) if share_logs: job_info.release_logs() diff --git a/packages/syft-job/tests/test_job_flow.py b/packages/syft-job/tests/test_job_flow.py index 019f27db871..f9bd965dd9a 100644 --- a/packages/syft-job/tests/test_job_flow.py +++ b/packages/syft-job/tests/test_job_flow.py @@ -93,7 +93,6 @@ def test_full_job_lifecycle(tmp_path: Path): assert job.status == "done" # --- Check outputs --- - # output_paths includes syft.pub.yaml created by _prepare_outputs_dir output_names = {p.name for p in job.output_paths} assert "result.txt" in output_names result_file = next(p for p in job.output_paths if p.name == "result.txt") @@ -109,7 +108,7 @@ def test_full_job_lifecycle(tmp_path: Path): returncode_path = staging_path / "returncode.txt" assert returncode_path.read_text().strip() == "0" # Nothing readable sits in review/ before the release. - for name in LOG_FILES: + for name in (*LOG_FILES, "outputs"): assert not (review_path / name).exists() # The viewer finds the staged file, because the DO reads both directories. assert "hello from job" in str(job.stdout) @@ -128,13 +127,12 @@ def test_full_job_lifecycle(tmp_path: Path): f"app_data/job/staging/{DS_EMAIL}/{PROTOCOL_DIR}/test.job/{name}" ).has_read_access(DS_EMAIL) - # --- Release the logs, then share outputs and logs with DS --- - job.share_outputs([DS_EMAIL]) + # --- Release the outputs and the logs to the DS --- + assert job.release(["outputs"], [DS_EMAIL]) == ["outputs"] assert sorted(job.release_logs()) == sorted(LOG_FILES) - job.share_logs([DS_EMAIL]) # The release moved the files into review/, which is the disclosure. - for name in LOG_FILES: + for name in (*LOG_FILES, "outputs"): assert (review_path / name).exists() assert not (staging_path / name).exists() diff --git a/packages/syft-job/tests/test_staging.py b/packages/syft-job/tests/test_staging.py index a0271a1e156..a2cd84f5dfe 100644 --- a/packages/syft-job/tests/test_staging.py +++ b/packages/syft-job/tests/test_staging.py @@ -135,7 +135,7 @@ def test_failure_after_run_stages_return_code(tmp_path, monkeypatch): def fail(*args, **kwargs): raise OSError("disk full") - monkeypatch.setattr(SyftJobRunner, "_move_outputs_to_review", fail) + monkeypatch.setattr(SyftJobRunner, "_move_outputs_to_staging", fail) job, review, staging = run_job(tmp_path, OK_PY, share_logs=None) assert (staging / "returncode.txt").read_text().strip() == "-1" diff --git a/tests/unit/test_sync_manager.py b/tests/unit/test_sync_manager.py index 9554379c1f5..a9d25289b60 100644 --- a/tests/unit/test_sync_manager.py +++ b/tests/unit/test_sync_manager.py @@ -505,7 +505,7 @@ def test_job_files_sync_to_submitter_only(): When a DO has multiple approved peers, job results should only be sent to the peer who submitted that specific job, not broadcast to all peers. - Uses permission-based routing: submitter gets read access via share_outputs(), + Uses permission-based routing: the outputs release grants the submitter read, non-submitter has no read access to job outputs. """ ds_manager, do_manager = SyftboxManager.pair_with_mock_drive_service_connection( From f0889cf7482237e581686e60e389c65d0be38ab1 Mon Sep 17 00:00:00 2001 From: Pedro Werneck Date: Fri, 30 Oct 2026 12:29:48 -0300 Subject: [PATCH 2/6] feat: release enclave results and disclosures through JobInfo.release The enclave releases disclosures to the submitter and the DOs, and pushes them to the same readers, so each party that gets a file can read it. Outputs go out through _release_results, which releases them and then pushes them to the same readers. A release with no readers moves nothing, so blocked outputs stay in staging/. The DO tutorial points to share_job_results instead of the removed share_outputs. --- docs/jobs.md | 2 +- .../ai_audit/internal/DO_Tutorial_V3.ipynb | 2 +- .../syft-enclave/src/syft_enclaves/client.py | 36 ++++++++++--------- .../syft-enclave/tests/test_disclosures.py | 13 +++++++ packages/syft-job/src/syft_job/job.py | 9 +++-- packages/syft-job/tests/test_policies.py | 3 +- 6 files changed, 43 insertions(+), 22 deletions(-) diff --git a/docs/jobs.md b/docs/jobs.md index fe6d585de55..b86bee9049f 100644 --- a/docs/jobs.md +++ b/docs/jobs.md @@ -261,7 +261,7 @@ An empty list holds the outputs back, but the job does not fail. The DO can shar ### What the DS sees -The DS always sees the job's status. When the output policy denies, the job is `failed` and the DS reads the reason in `job.policy_error`. When the policy removed the DS from the recipients, the job shows "Outputs not shared with you". Blocked outputs stay in the job's review folder, where only the DO reads them. +The DS always sees the job's status. When the output policy denies, the job is `failed` and the DS reads the reason in `job.policy_error`. When the policy removed the DS from the recipients, the job shows "Outputs not shared with you". Blocked outputs stay in the job's staging folder, where only the DO reads them. ### Example: block a word diff --git a/notebooks/ai_audit/internal/DO_Tutorial_V3.ipynb b/notebooks/ai_audit/internal/DO_Tutorial_V3.ipynb index f6f846222ba..2a0ccf13050 100644 --- a/notebooks/ai_audit/internal/DO_Tutorial_V3.ipynb +++ b/notebooks/ai_audit/internal/DO_Tutorial_V3.ipynb @@ -791,7 +791,7 @@ "\n", "By default, results stay on your machine. To send them back to the DS, pass `share_outputs_with_submitter=True`. Almost always what you want.\n", "\n", - "> 💡 Alternative: leave it `False` for jobs where you want to inspect the outputs first, then call `job.share_outputs([ds_email])` manually after review." + "> 💡 Alternative: leave it `False` for jobs where you want to inspect the outputs first, then call `client.job_runner.share_job_results(job.name, share_outputs=True, share_logs=False)` after review." ] }, { diff --git a/packages/syft-enclave/src/syft_enclaves/client.py b/packages/syft-enclave/src/syft_enclaves/client.py index d5fc786a025..a5398d12602 100644 --- a/packages/syft-enclave/src/syft_enclaves/client.py +++ b/packages/syft-enclave/src/syft_enclaves/client.py @@ -474,15 +474,25 @@ def _local_jobs(self) -> JobsList: return self._rds.job_client.jobs def _apply_disclosure_policy(self) -> None: - """Move each granted artifact from the staging folder into review. + """Release each granted artifact to the parties of its job. An artifact that no party released stays in the staging folder, where no - submitter holds a grant. + party holds a grant. """ for job in self._local_jobs(): if job.status not in ("done", "failed"): continue - job.release_artifacts(gated_names(self.granted_disclosures(job))) + job.release( + gated_names(self.granted_disclosures(job)), + self._disclosure_readers(job), + ) + + def _disclosure_readers(self, job: JobInfo) -> list[str]: + """The submitter and each DO of the job, who all get a released artifact.""" + datasets = job.job_metadata.datasets or {} + return list( + dict.fromkeys([job.submitted_by, *(e for e in datasets if e != self.email)]) + ) def distribute_results(self) -> None: """Distribute job results to DS (always) and optionally to DOs.""" @@ -506,15 +516,12 @@ def _share_results_once(self, job: JobInfo) -> None: self._try_write_receipt(job) # Always share results with the DS (submitter) - job.release(["outputs"], [job.submitted_by]) - self._forward_results_to_recipients(job, [job.submitted_by]) + self._release_results(job, [job.submitted_by]) # Optionally share with DOs datasets = job.job_metadata.datasets if job.job_headers.get("share_results_with_do") and datasets: - do_emails = list(datasets.keys()) - job.release(["outputs"], do_emails) - self._forward_results_to_recipients(job, do_emails) + self._release_results(job, list(datasets.keys())) marker.write_text("shared") @@ -547,11 +554,7 @@ def _forward_new_releases(self, job: JobInfo) -> list[str]: ).read_bytes() for name in pending } - datasets = job.job_metadata.datasets or {} - recipients = list( - dict.fromkeys([job.submitted_by, *(e for e in datasets if e != self.email)]) - ) - self._push_files_to_recipients(files, recipients) + self._push_files_to_recipients(files, self._disclosure_readers(job)) sent_path.write_text(json.dumps(sorted(sent.union(pending)))) return pending @@ -577,14 +580,15 @@ def _read_state_file(self, job: JobInfo) -> dict[Path, bytes]: state_rel = state_file.relative_to(datasite_dir) return {state_rel: state_file.read_bytes()} - def _forward_results_to_recipients(self, job: JobInfo, recipients: list[str]): - """Forward job output files and state to recipients via event outbox.""" + def _release_results(self, job: JobInfo, readers: list[str]) -> None: + """Release the outputs to ``readers``, and push them with the job state.""" + job.release(["outputs"], readers) outputs_dir = job.job_review_path / "outputs" files_by_datasite_path = ( self._get_files_in_dir(outputs_dir) if outputs_dir.exists() else {} ) files_by_datasite_path.update(self._read_state_file(job)) - self._push_files_to_recipients(files_by_datasite_path, recipients) + self._push_files_to_recipients(files_by_datasite_path, readers) def _push_files_to_recipients( self, files_by_datasite_path: dict, recipients: list[str] diff --git a/packages/syft-enclave/tests/test_disclosures.py b/packages/syft-enclave/tests/test_disclosures.py index 4245897e303..5eff2d957f9 100644 --- a/packages/syft-enclave/tests/test_disclosures.py +++ b/packages/syft-enclave/tests/test_disclosures.py @@ -17,6 +17,7 @@ from syft_job.job import JobInfo # noqa: E402 from syft_job.models import JobState, JobStatus # noqa: E402 from syft_job.traceback_capture import FRAMES_FILENAME # noqa: E402 +from syft_perms import SyftPermContext # noqa: E402 from syft_enclaves import SyftEnclaveClient # noqa: E402 from syft_enclaves.enclave_job_info import ( # noqa: E402 @@ -249,6 +250,18 @@ def test_frames_reach_data_owners(): assert (Path(do1.jobs["j"].job_review_path) / FRAMES_FILENAME).exists() +def test_released_artifact_grants_read_to_parties(): + enclave, do1, do2, ds = build_quad() + submit(ds, enclave, do1, do2, CRASH_CODE, [FRAMES]) + run_to_completion(enclave, do1, do2, ds, [FRAMES]) + + datasite = enclave._rds.syftbox_folder / enclave.email + frames = Path(enclave.jobs["j"].job_review_path) / FRAMES_FILENAME + ctx = SyftPermContext(datasite=datasite) + for party in (ds, do1, do2): + assert ctx.open(str(frames.relative_to(datasite))).has_read_access(party.email) + + def test_one_owner_withholding_blocks_frames(): enclave, do1, do2, ds = build_quad() submit(ds, enclave, do1, do2, CRASH_CODE, [FRAMES]) diff --git a/packages/syft-job/src/syft_job/job.py b/packages/syft-job/src/syft_job/job.py index 7533d9de36e..97c63ca0a52 100644 --- a/packages/syft-job/src/syft_job/job.py +++ b/packages/syft-job/src/syft_job/job.py @@ -107,10 +107,13 @@ def release_logs(self) -> list[str]: def release(self, filenames: Iterable[str], readers: list[str]) -> list[str]: """Move the named staged files to review/, and give ``readers`` read access. - Returns the names that moved. A file that is already in review/ also - gets the read access, so a caller can add a reader later. A reader with - read access on the whole review folder also sees each file that moves. + Returns the names that moved. With no ``readers``, no file moves. A file + that is already in review/ also gets the read access, so a caller can + add a reader later. A reader with read access on the whole review + folder also sees each file that moves. """ + if not readers: + return [] filenames = list(filenames) moved = self.release_artifacts(filenames) released = [n for n in filenames if (self.job_review_path / n).exists()] diff --git a/packages/syft-job/tests/test_policies.py b/packages/syft-job/tests/test_policies.py index 4d51897b716..d1c53906c85 100644 --- a/packages/syft-job/tests/test_policies.py +++ b/packages/syft-job/tests/test_policies.py @@ -204,7 +204,7 @@ def test_output_policy_blocks_word_and_keeps_outputs_with_do(env): assert job.status == "failed" assert job.policy_error == "ValueError: Output out.txt contains a blocked word" assert env.reads(DS_EMAIL, "call.job", "policy_error.txt") - assert (job.job_review_path / "outputs" / "out.txt").read_text() == "SECRET\n" + assert (job.job_staging_path / "outputs" / "out.txt").read_text() == "SECRET\n" assert not env.reads(DS_EMAIL, "call.job", "outputs/out.txt") @@ -246,6 +246,7 @@ def test_empty_recipients_hold_outputs_back(env): assert job.status == "done" assert not env.reads(DS_EMAIL, "call.job", "outputs/out.txt") assert job.outputs_withheld and job.policy_error is None + assert not (job.job_review_path / "outputs").exists() assert "Outputs not shared with you" in _ds_job(env)._repr_html_() env.runner.share_job_results("call.job", True, False, user=DS_EMAIL) From 3c996a924476c846794cb39cbadf7c66eb3f2c28 Mon Sep 17 00:00:00 2001 From: Pedro Werneck Date: Fri, 30 Oct 2026 13:17:55 -0300 Subject: [PATCH 3/6] refactor: name job result sharing by release and recipients release_artifacts becomes the private _move_to_review, so release is the only public way to release a file. The private share and forward methods become _release_job_results, _release_results_once, _release_outputs and _push_new_disclosures, and the people who get a file are recipients throughout. JobStorage.remove_outputs replaces the two loops that deleted the outputs of an earlier run. --- .../syft-enclave/src/syft_enclaves/client.py | 27 +++++++------ .../syft-enclave/tests/test_disclosures.py | 2 +- packages/syft-job/src/syft_job/job.py | 40 +++++++++---------- packages/syft-job/src/syft_job/job_runner.py | 18 ++++----- packages/syft-job/src/syft_job/job_storage.py | 15 ++++++- .../syft-job/src/syft_job/policies/models.py | 2 +- packages/syft-job/tests/test_staging.py | 15 +++---- 7 files changed, 65 insertions(+), 54 deletions(-) diff --git a/packages/syft-enclave/src/syft_enclaves/client.py b/packages/syft-enclave/src/syft_enclaves/client.py index a5398d12602..76589602d5c 100644 --- a/packages/syft-enclave/src/syft_enclaves/client.py +++ b/packages/syft-enclave/src/syft_enclaves/client.py @@ -484,10 +484,10 @@ def _apply_disclosure_policy(self) -> None: continue job.release( gated_names(self.granted_disclosures(job)), - self._disclosure_readers(job), + self._disclosure_recipients(job), ) - def _disclosure_readers(self, job: JobInfo) -> list[str]: + def _disclosure_recipients(self, job: JobInfo) -> list[str]: """The submitter and each DO of the job, who all get a released artifact.""" datasets = job.job_metadata.datasets or {} return list( @@ -499,14 +499,14 @@ def distribute_results(self) -> None: for job in self.jobs: if job.status not in ("done", "failed"): continue - self._share_results_once(job) + self._release_results_once(job) # A grant can arrive after the results went out, so a released # artifact travels on its own schedule. - self._forward_new_releases(job) + self._push_new_disclosures(job) self._rds.sync() - def _share_results_once(self, job: JobInfo) -> None: + def _release_results_once(self, job: JobInfo) -> None: """Send the outputs and the state, the first time the job finishes.""" marker = job.job_review_path / RESULTS_SHARED_MARKER if marker.exists(): @@ -516,16 +516,17 @@ def _share_results_once(self, job: JobInfo) -> None: self._try_write_receipt(job) # Always share results with the DS (submitter) - self._release_results(job, [job.submitted_by]) + self._release_outputs(job, [job.submitted_by]) # Optionally share with DOs datasets = job.job_metadata.datasets if job.job_headers.get("share_results_with_do") and datasets: - self._release_results(job, list(datasets.keys())) + do_emails = list(datasets) + self._release_outputs(job, do_emails) marker.write_text("shared") - def _forward_new_releases(self, job: JobInfo) -> list[str]: + def _push_new_disclosures(self, job: JobInfo) -> list[str]: """Send each released artifact that the parties do not hold yet. A party that releases an item also receives it, so the data owners get @@ -554,7 +555,7 @@ def _forward_new_releases(self, job: JobInfo) -> list[str]: ).read_bytes() for name in pending } - self._push_files_to_recipients(files, self._disclosure_readers(job)) + self._push_files_to_recipients(files, self._disclosure_recipients(job)) sent_path.write_text(json.dumps(sorted(sent.union(pending)))) return pending @@ -580,15 +581,15 @@ def _read_state_file(self, job: JobInfo) -> dict[Path, bytes]: state_rel = state_file.relative_to(datasite_dir) return {state_rel: state_file.read_bytes()} - def _release_results(self, job: JobInfo, readers: list[str]) -> None: - """Release the outputs to ``readers``, and push them with the job state.""" - job.release(["outputs"], readers) + def _release_outputs(self, job: JobInfo, recipients: list[str]) -> None: + """Release the outputs to ``recipients``, and push them with the job state.""" + job.release(["outputs"], recipients) outputs_dir = job.job_review_path / "outputs" files_by_datasite_path = ( self._get_files_in_dir(outputs_dir) if outputs_dir.exists() else {} ) files_by_datasite_path.update(self._read_state_file(job)) - self._push_files_to_recipients(files_by_datasite_path, readers) + self._push_files_to_recipients(files_by_datasite_path, recipients) def _push_files_to_recipients( self, files_by_datasite_path: dict, recipients: list[str] diff --git a/packages/syft-enclave/tests/test_disclosures.py b/packages/syft-enclave/tests/test_disclosures.py index 5eff2d957f9..d3b5c99e6f2 100644 --- a/packages/syft-enclave/tests/test_disclosures.py +++ b/packages/syft-enclave/tests/test_disclosures.py @@ -379,7 +379,7 @@ def test_released_artifact_not_sent_twice(): run_to_completion(enclave, do1, do2, ds, [FRAMES]) job = enclave.jobs["j"] - assert enclave._forward_new_releases(job) == [] + assert enclave._push_new_disclosures(job) == [] def test_approve_job_rejects_non_enclave_job(): diff --git a/packages/syft-job/src/syft_job/job.py b/packages/syft-job/src/syft_job/job.py index 97c63ca0a52..b26689d7255 100644 --- a/packages/syft-job/src/syft_job/job.py +++ b/packages/syft-job/src/syft_job/job.py @@ -104,23 +104,23 @@ def release_logs(self) -> list[str]: """ return self.release(STAGED_LOG_FILES, [self.submitted_by]) - def release(self, filenames: Iterable[str], readers: list[str]) -> list[str]: - """Move the named staged files to review/, and give ``readers`` read access. + def release(self, filenames: Iterable[str], recipients: list[str]) -> list[str]: + """Move the named staged files to review/, for ``recipients`` to read. - Returns the names that moved. With no ``readers``, no file moves. A file - that is already in review/ also gets the read access, so a caller can - add a reader later. A reader with read access on the whole review + Returns the names that moved. With no ``recipients``, no file moves. A + file that is already in review/ also gets the read access, so a caller + can add a recipient later. A user with read access on the whole review folder also sees each file that moves. """ - if not readers: + if not recipients: return [] filenames = list(filenames) - moved = self.release_artifacts(filenames) + moved = self._move_to_review(filenames) released = [n for n in filenames if (self.job_review_path / n).exists()] - self._grant_read(released, readers) + self._grant_read(released, recipients) return moved - def release_artifacts(self, filenames: Iterable[str]) -> list[str]: + def _move_to_review(self, filenames: Iterable[str]) -> list[str]: """Move the named staged files to review/, and give no read access. Returns the names that moved. The method skips a name that is not in @@ -210,7 +210,10 @@ def update_disclosures(self, disclosures: DisclosuresArg) -> dict[str, bool]: return granted def release_disclosures(self) -> list[str]: - """Release the granted items to the submitter, and return the names moved.""" + """Move the granted items to review/, for the submitter to read. + + Returns the names that moved. + """ return self.release(gated_names(self.granted_disclosures), [self.submitted_by]) # ────────────────────────────────────────────── @@ -614,13 +617,8 @@ def rerun(self) -> None: f.unlink() changes_made.append(filename) - for outputs_dir in ( - self.job_review_path / "outputs", - self.job_staging_path / "outputs", - ): - if outputs_dir.is_dir(): - shutil.rmtree(outputs_dir) - changes_made.append("outputs directory") + if self._client.manager.remove_outputs(self._ref): + changes_made.append("outputs directory") # Reset state to approved self._state.status = JobStatus.APPROVED @@ -652,13 +650,13 @@ def _relative_review_path(self, subpath: str) -> Path: ) return rel / subpath - def _grant_read(self, filenames: Iterable[str], users: list[str]) -> None: - """Grant read access to the named review/ files for the given users.""" + def _grant_read(self, filenames: Iterable[str], recipients: list[str]) -> None: + """Grant ``recipients`` read access to the named review/ files.""" ctx = self._get_perm_context() for filename in filenames: f = ctx.open(self._relative_review_path(filename)) - for user in users: - f.grant_read_access(user) + for recipient in recipients: + f.grant_read_access(recipient) # ────────────────────────────────────────────── # Display diff --git a/packages/syft-job/src/syft_job/job_runner.py b/packages/syft-job/src/syft_job/job_runner.py index 9abfa90579d..3c1c11b45d2 100644 --- a/packages/syft-job/src/syft_job/job_runner.py +++ b/packages/syft-job/src/syft_job/job_runner.py @@ -755,12 +755,7 @@ def _prepare_outputs_dir(self, ref: JobRef, run_dir: Path) -> None: # Create outputs/ inside code/ dir so job scripts can write there (cwd is code/) (run_dir / "code" / "outputs").mkdir(parents=True, exist_ok=True) - for outputs_dir in ( - self.manager.review_dir(ref) / "outputs", - self.manager.staging_dir(ref) / "outputs", - ): - if outputs_dir.exists(): - shutil.rmtree(outputs_dir) + self.manager.remove_outputs(ref) def _replace_ds_folder_reads(self) -> None: """Replace the read grant on a whole DS review folder with the named files. @@ -859,7 +854,10 @@ def process_approved_jobs( does not identify a job — it is unique per datasite and submitter — so skipping by name drops every other job that shares it. - share_outputs_with_submitter: If True, grant read access on outputs to submitter. + share_outputs_with_submitter: If True, release the outputs to the + output recipients: the submitter, or the list an output policy + writes. False keeps them in staging, where only the DO reads + them. share_logs_with_submitter: If True, release the logs and the exit code to the submitter, whatever the job requested. False (default) releases only the items that the submitter requested @@ -889,7 +887,7 @@ def process_approved_jobs( for ref in approved_jobs: print(f"\n{'=' * 50}") if self._execute_job(ref, stream_output=stream_output, timeout=timeout): - self._share_job_results( + self._release_job_results( ref, share_outputs_with_submitter, share_logs_with_submitter, @@ -919,9 +917,9 @@ def share_job_results( (self.manager.review_dir(ref) / OUTPUTS_WITHHELD_FILENAME).unlink( missing_ok=True ) - self._share_job_results(ref, share_outputs, share_logs, [ref.ds_email]) + self._release_job_results(ref, share_outputs, share_logs, [ref.ds_email]) - def _share_job_results( + def _release_job_results( self, ref: JobRef, share_outputs: bool, share_logs: bool, recipients: list[str] ) -> None: job_info = self._get_job_info(ref) diff --git a/packages/syft-job/src/syft_job/job_storage.py b/packages/syft-job/src/syft_job/job_storage.py index edc9ca4c26c..ac7c087fc73 100644 --- a/packages/syft-job/src/syft_job/job_storage.py +++ b/packages/syft-job/src/syft_job/job_storage.py @@ -1,4 +1,5 @@ import logging +import shutil from pathlib import Path from typing import Callable, Iterator, Optional @@ -16,8 +17,8 @@ from .models import JobState, JobSubmissionMetadata, RequestMetadata from .protocolcodecs import CODECS, ProtocolCodec from .submission import ( - REQUEST_FILE_NAME, OWNER_RECORD_FILENAME, + REQUEST_FILE_NAME, OwnerJobRecord, claim_owner_record, read_owner_record, @@ -135,6 +136,18 @@ def review_dir(self, ref: JobRef) -> Path: def staging_dir(self, ref: JobRef) -> Path: return self._codec_for(ref.protocol_version).staging_dir(ref) + def remove_outputs(self, ref: JobRef) -> bool: + """Delete outputs/ in review/ and in staging/. Returns True if one existed.""" + removed = False + for outputs_dir in ( + self.review_dir(ref) / "outputs", + self.staging_dir(ref) / "outputs", + ): + if outputs_dir.is_dir(): + shutil.rmtree(outputs_dir) + removed = True + return removed + # -- owner record (owner side, in staging/) ------------------------------------ def owner_record_path(self, ref: JobRef) -> Path: return self.staging_dir(ref) / OWNER_RECORD_FILENAME diff --git a/packages/syft-job/src/syft_job/policies/models.py b/packages/syft-job/src/syft_job/policies/models.py index 4b19d5719a5..549e4aeb262 100644 --- a/packages/syft-job/src/syft_job/policies/models.py +++ b/packages/syft-job/src/syft_job/policies/models.py @@ -10,7 +10,7 @@ # Job artifacts. policy_error.txt and outputs_withheld.txt are released to the # submitter; the rest stays in staging, with the data owner. POLICY_ERROR_FILENAME = "policy_error.txt" -# An empty marker file; see SyftJobRunner._share_job_results(). +# An empty marker file; see SyftJobRunner._release_job_results(). OUTPUTS_WITHHELD_FILENAME = "outputs_withheld.txt" OUTPUT_RECIPIENTS_FILENAME = "output_recipients.json" diff --git a/packages/syft-job/tests/test_staging.py b/packages/syft-job/tests/test_staging.py index a2cd84f5dfe..6d69b68a588 100644 --- a/packages/syft-job/tests/test_staging.py +++ b/packages/syft-job/tests/test_staging.py @@ -6,22 +6,20 @@ import pytest import yaml -from syft_permissions.spec.ruleset import PERMISSION_FILE_NAME -from syft_perms import SyftPermContext - from syft_job.client import JobClient -from syft_job.config import SyftJobConfig +from syft_job.config import SyftJobConfig, protocol_dir_name from syft_job.disclosures import LOGS_WARNING, DisclosureItem from syft_job.job import JobInfo from syft_job.job_runner import SyftJobRunner +from syft_job.migrations.registry import JOB_PROTOCOL_VERSION from syft_job.models import JobState, JobStatus from syft_job.submission import ( OWNER_RECORD_FILENAME, SUBMISSION_HASH_HEADER, ) from syft_job.traceback_capture import FRAMES_FILENAME -from syft_job.config import protocol_dir_name -from syft_job.migrations.registry import JOB_PROTOCOL_VERSION +from syft_permissions.spec.ruleset import PERMISSION_FILE_NAME +from syft_perms import SyftPermContext # The path segment of the current job protocol, e.g. "v2". PROTOCOL_DIR = protocol_dir_name(JOB_PROTOCOL_VERSION) @@ -196,7 +194,7 @@ def test_traceback_record_is_staged(tmp_path): assert record["chain"][0]["type"] == "ValueError" assert "4120550" not in record_path.read_text() - assert job.release_artifacts([FRAMES_FILENAME]) == [FRAMES_FILENAME] + assert job.release([FRAMES_FILENAME], [DS_EMAIL]) == [FRAMES_FILENAME] assert (review / FRAMES_FILENAME).exists() @@ -205,12 +203,15 @@ def test_rerun_clears_both_locations(tmp_path): job, review, staging = run_job(tmp_path, OK_PY, share_logs=False) (review / "returncode.txt").write_text("0") assert (staging / "stdout.txt").exists() + assert (review / "outputs").exists() or (staging / "outputs").exists() job.rerun() assert not (staging / "stdout.txt").exists() assert not (staging / "stderr.txt").exists() assert not (review / "stdout.txt").exists() + assert not (review / "outputs").exists() + assert not (staging / "outputs").exists() def test_rerun_clears_crash_record(tmp_path): From 3a40c141dfec190a67a26793e06bfc8e25aa9a5f Mon Sep 17 00:00:00 2001 From: Pedro Werneck Date: Fri, 30 Oct 2026 14:31:49 -0300 Subject: [PATCH 4/6] fix: guard staged job outputs against permission files and symlinks A syft.pub.yaml in the staged outputs decided who reads them, so a job could release its own outputs past the output policy. The runner now fails such a job, shares its outputs with nobody, and records the reason in staging/ for the DO only. Each run also drops the recipients and the outputs error of the run before. shutil.move kept the job's symlinks, so a link into the run copy broke when the run copy was removed, an absolute link stayed live in the SyftBox, and a symlinked outputs folder made rerun fail. The outputs are now copied with the links followed, as the runner did before outputs were staged, and the copy skips the permission files. --- packages/syft-job/src/syft_job/job_runner.py | 65 +++++++++++++-- packages/syft-job/tests/test_staging.py | 84 +++++++++++++++++++- 2 files changed, 139 insertions(+), 10 deletions(-) diff --git a/packages/syft-job/src/syft_job/job_runner.py b/packages/syft-job/src/syft_job/job_runner.py index 3c1c11b45d2..0122ac62346 100644 --- a/packages/syft-job/src/syft_job/job_runner.py +++ b/packages/syft-job/src/syft_job/job_runner.py @@ -11,6 +11,7 @@ from typing import List, Set import psutil +from syft_permissions.spec.ruleset import PERMISSION_FILE_NAME from . import __version__ from .client import JobClient @@ -22,10 +23,10 @@ OUTPUT_RECIPIENTS_FILENAME, OUTPUTS_WITHHELD_FILENAME, POLICY_ERROR_FILENAME, + ApiResolver, JobPolicies, PolicyEnv, PolicyKind, - ApiResolver, PolicyResolutionError, PolicyResult, PolicyRunner, @@ -105,6 +106,9 @@ def _format_timeout(seconds: int) -> str: "approved without a record of the approved submission; review and approve it again" ) CHANGED_SUBMISSION_REASON = "the submission changed after it was approved" +# Why the runner failed a job for its outputs. No release names it, so only +# the DO reads it. +OUTPUTS_ERROR_FILENAME = "outputs_error.txt" logger = logging.getLogger(__name__) @@ -614,7 +618,8 @@ def _run_approved_job( self._capture_traceback(ref, returncode, run_dir / "code") returncode = self._run_output_policy(ref, run_dir, policies, returncode) - self._move_outputs_to_staging(ref, run_dir) + skipped = self._copy_outputs_to_staging(ref, run_dir) + returncode = self._fail_on_output_permission_files(ref, skipped, returncode) self._finalize(ref, returncode) self._report_result(ref, returncode) @@ -740,22 +745,66 @@ def _finalize(self, ref: JobRef, returncode: int) -> None: state.return_code = None self.manager.write_state(ref, state) - def _move_outputs_to_staging(self, ref: JobRef, run_dir: Path) -> None: - """Move the outputs from the run copy to staging/, to wait for a release.""" - staging_dir = self.manager.staging_dir(ref) - staging_dir.mkdir(parents=True, exist_ok=True) - shutil.move(str(run_dir / "code" / "outputs"), str(staging_dir / "outputs")) + def _fail_on_output_permission_files( + self, ref: JobRef, permission_files: list[str], returncode: int + ) -> int: + """Fail the job when it wrote a permission file into outputs/. + + In staging/, that file would decide who reads the outputs, so the job + could release them itself. The copy to staging/ skips the file, the job + shares its outputs with nobody, and the reason goes to staging/ for the + DO. Returns the exit code to record. + """ + if not permission_files: + return returncode + reason = ( + f"Job {ref.job_name} wrote permission files into outputs/: " + f"{', '.join(permission_files)}. The job failed and shares its " + f"outputs with nobody." + ) + logger.error(reason) + self._write_staged(ref, OUTPUTS_ERROR_FILENAME, f"{reason}\n") + self._write_recipients(ref, []) + return returncode or 1 + + def _copy_outputs_to_staging(self, ref: JobRef, run_dir: Path) -> list[str]: + """Copy the outputs from the run copy to staging/, to wait for a release. + + The copy follows symlinks, so staging/ holds only real files. It skips + each permission file, and returns their paths in outputs/. + """ + run_outputs = run_dir / "code" / "outputs" + skipped: list[str] = [] + + def skip_permission_files(folder: str, names: list[str]) -> set[str]: + if PERMISSION_FILE_NAME not in names: + return set() + relative = (Path(folder) / PERMISSION_FILE_NAME).relative_to(run_outputs) + skipped.append(relative.as_posix()) + return {PERMISSION_FILE_NAME} + + shutil.copytree( + run_outputs, + self.manager.staging_dir(ref) / "outputs", + ignore=skip_permission_files, + ) + return sorted(skipped) def _prepare_outputs_dir(self, ref: JobRef, run_dir: Path) -> None: """Create outputs/ in the run copy, and remove the outputs of a previous run. The job writes its outputs in the run copy, outside the SyftBox, so no - sync can send them to the submitter before they are released. + sync can send them to the submitter before they are released. The + recipients and the outputs error of a previous run go too, or this run + inherits them. """ # Create outputs/ inside code/ dir so job scripts can write there (cwd is code/) (run_dir / "code" / "outputs").mkdir(parents=True, exist_ok=True) self.manager.remove_outputs(ref) + staging_dir = self.manager.staging_dir(ref) + for filename in (OUTPUT_RECIPIENTS_FILENAME, OUTPUTS_ERROR_FILENAME): + (staging_dir / filename).unlink(missing_ok=True) def _replace_ds_folder_reads(self) -> None: """Replace the read grant on a whole DS review folder with the named files. diff --git a/packages/syft-job/tests/test_staging.py b/packages/syft-job/tests/test_staging.py index 6d69b68a588..c0851dc3fea 100644 --- a/packages/syft-job/tests/test_staging.py +++ b/packages/syft-job/tests/test_staging.py @@ -10,9 +10,10 @@ from syft_job.config import SyftJobConfig, protocol_dir_name from syft_job.disclosures import LOGS_WARNING, DisclosureItem from syft_job.job import JobInfo -from syft_job.job_runner import SyftJobRunner +from syft_job.job_runner import OUTPUTS_ERROR_FILENAME, SyftJobRunner from syft_job.migrations.registry import JOB_PROTOCOL_VERSION from syft_job.models import JobState, JobStatus +from syft_job.policies import OUTPUT_RECIPIENTS_FILENAME from syft_job.submission import ( OWNER_RECORD_FILENAME, SUBMISSION_HASH_HEADER, @@ -133,7 +134,7 @@ def test_failure_after_run_stages_return_code(tmp_path, monkeypatch): def fail(*args, **kwargs): raise OSError("disk full") - monkeypatch.setattr(SyftJobRunner, "_move_outputs_to_staging", fail) + monkeypatch.setattr(SyftJobRunner, "_copy_outputs_to_staging", fail) job, review, staging = run_job(tmp_path, OK_PY, share_logs=None) assert (staging / "returncode.txt").read_text().strip() == "-1" @@ -229,6 +230,85 @@ def test_rerun_clears_crash_record(tmp_path): assert not (review / FRAMES_FILENAME).exists() +PERMISSION_FILE_PY = f"""\ +import os + +os.makedirs("outputs", exist_ok=True) +with open("outputs/result.txt", "w") as f: + f.write("done") +with open("outputs/{PERMISSION_FILE_NAME}", "w") as f: + f.write("rules:\\n- pattern: '**'\\n access:\\n read: ['*']\\n") +""" + + +def test_permission_file_in_outputs_fails_job(tmp_path): + """A job must not decide who reads its own staged outputs.""" + job, _, staging = run_job(tmp_path, PERMISSION_FILE_PY, share_logs=None) + + assert job.status == "failed" + assert (staging / "outputs" / "result.txt").exists() + assert not (staging / "outputs" / PERMISSION_FILE_NAME).exists() + ctx = SyftPermContext(datasite=tmp_path / "SyftBox" / DO_EMAIL) + staged_result = ( + f"app_data/job/staging/{DS_EMAIL}/{PROTOCOL_DIR}/test.job/outputs/result.txt" + ) + assert not ctx.open(staged_result).has_read_access(DS_EMAIL) + assert PERMISSION_FILE_NAME in (staging / OUTPUTS_ERROR_FILENAME).read_text() + + +def test_symlinked_outputs_staged_as_files(tmp_path): + external = tmp_path / "external.txt" + external.write_text("external") + code = f"""\ +import os + +os.makedirs("outputs", exist_ok=True) +with open("checkpoint.txt", "w") as f: + f.write("checkpoint") +os.symlink("../checkpoint.txt", "outputs/latest.txt") +os.symlink({str(external)!r}, "outputs/external.txt") +""" + _, _, staging = run_job(tmp_path, code, share_logs=None) + + for name, content in (("latest.txt", "checkpoint"), ("external.txt", "external")): + staged = staging / "outputs" / name + assert not staged.is_symlink() + assert staged.read_text() == content + + +def test_symlinked_outputs_folder_staged_as_folder(tmp_path): + results = tmp_path / "results" + results.mkdir() + (results / "result.txt").write_text("done") + code = f"""\ +import os +import shutil + +shutil.rmtree("outputs", ignore_errors=True) +os.symlink({str(results)!r}, "outputs") +""" + job, _, staging = run_job(tmp_path, code, share_logs=None) + + assert not (staging / "outputs").is_symlink() + assert (staging / "outputs" / "result.txt").read_text() == "done" + job.rerun() + assert not (staging / "outputs").exists() + + +def test_new_run_drops_recipients_and_outputs_error_of_old_run(tmp_path): + """A run that blocked its outputs must not block the outputs of the next.""" + job, _, staging = run_job(tmp_path, OK_PY, share_logs=None) + (staging / OUTPUT_RECIPIENTS_FILENAME).write_text("[]") + (staging / OUTPUTS_ERROR_FILENAME).write_text("old run\n") + + job.rerun() + run_approved(tmp_path) + + assert reload_job(tmp_path).status == "done" + assert not (staging / OUTPUT_RECIPIENTS_FILENAME).exists() + assert not (staging / OUTPUTS_ERROR_FILENAME).exists() + + # -- disclosures: requested by the DS, granted by the DO ------------------------ From 74cd93cd0532503048c44288fb84ddc45ace70b5 Mon Sep 17 00:00:00 2001 From: Pedro Werneck Date: Fri, 30 Oct 2026 15:27:36 -0300 Subject: [PATCH 5/6] fix: drop stale outputs grants and duplicate enclave sends remove_outputs also removes the read grant on the outputs, so the outputs of a rerun reach only the recipients of the new release. release() builds no permission context when nothing is released, and the enclave releases a job only while a granted item is still staged, so a tick with nothing new does no permission work. The enclave pushes only the job state. The read grant and the sync already deliver released outputs and disclosures, so the push sent them twice; forwarded_artifacts.json goes with it. The runner docstrings say outputs wait in staging/, and the rerun test now checks that the released outputs are removed. --- .../syft-enclave/src/syft_enclaves/client.py | 66 ++++--------------- .../syft-enclave/tests/test_disclosures.py | 40 ++++++++++- packages/syft-job/src/syft_job/job.py | 3 +- packages/syft-job/src/syft_job/job_runner.py | 8 +-- packages/syft-job/src/syft_job/job_storage.py | 13 +++- packages/syft-job/tests/test_staging.py | 37 ++++++++++- 6 files changed, 104 insertions(+), 63 deletions(-) diff --git a/packages/syft-enclave/src/syft_enclaves/client.py b/packages/syft-enclave/src/syft_enclaves/client.py index 76589602d5c..b5599eca51e 100644 --- a/packages/syft-enclave/src/syft_enclaves/client.py +++ b/packages/syft-enclave/src/syft_enclaves/client.py @@ -1,10 +1,10 @@ -import json import logging import os from datetime import datetime, timezone from pathlib import Path from typing import TYPE_CHECKING, Optional +from syft.sync.transport.protocols import Transport from syft_datasets import Dataset from syft_datasets.dataset_manager import SyftDatasetManager from syft_job.disclosures import DisclosuresArg, gated_names @@ -50,7 +50,6 @@ write_versions, ) -FORWARDED_RECORD = "forwarded_artifacts.json" RESULTS_SHARED_MARKER = "results_shared" # How the enclave records an approval: every data owner approved the same job. ENCLAVE_APPROVAL_METHOD = "enclave" @@ -477,15 +476,16 @@ def _apply_disclosure_policy(self) -> None: """Release each granted artifact to the parties of its job. An artifact that no party released stays in the staging folder, where no - party holds a grant. + party holds a grant. A job with no granted artifact left in staging is + skipped, because each release scans the datasite permissions. """ for job in self._local_jobs(): if job.status not in ("done", "failed"): continue - job.release( - gated_names(self.granted_disclosures(job)), - self._disclosure_recipients(job), - ) + granted = gated_names(self.granted_disclosures(job)) + if not any((job.job_staging_path / name).exists() for name in granted): + continue + job.release(granted, self._disclosure_recipients(job)) def _disclosure_recipients(self, job: JobInfo) -> list[str]: """The submitter and each DO of the job, who all get a released artifact.""" @@ -500,14 +500,11 @@ def distribute_results(self) -> None: if job.status not in ("done", "failed"): continue self._release_results_once(job) - # A grant can arrive after the results went out, so a released - # artifact travels on its own schedule. - self._push_new_disclosures(job) self._rds.sync() def _release_results_once(self, job: JobInfo) -> None: - """Send the outputs and the state, the first time the job finishes.""" + """Release the outputs and send the state, the first time the job finishes.""" marker = job.job_review_path / RESULTS_SHARED_MARKER if marker.exists(): return @@ -526,40 +523,6 @@ def _release_results_once(self, job: JobInfo) -> None: marker.write_text("shared") - def _push_new_disclosures(self, job: JobInfo) -> list[str]: - """Send each released artifact that the parties do not hold yet. - - A party that releases an item also receives it, so the data owners get - the same file as the submitter. The record names what already went out, - because a grant can arrive long after the results did. - """ - review_dir = job.job_review_path - sent_path = review_dir / FORWARDED_RECORD - try: - sent = set(json.loads(sent_path.read_text())) - except (OSError, TypeError, json.JSONDecodeError): - sent = set() - - pending = sorted( - name - for name in gated_names(self.granted_disclosures(job)) - if name not in sent and (review_dir / name).is_file() - ) - if not pending: - return [] - - datasite_dir = self._rds.syftbox_folder / self._rds.email - files = { - (review_dir / name).relative_to(datasite_dir): ( - review_dir / name - ).read_bytes() - for name in pending - } - self._push_files_to_recipients(files, self._disclosure_recipients(job)) - - sent_path.write_text(json.dumps(sorted(sent.union(pending)))) - return pending - def _try_write_receipt(self, job: JobInfo) -> None: """Sign a receipt into the job's outputs, so it ships with them. @@ -582,14 +545,13 @@ def _read_state_file(self, job: JobInfo) -> dict[Path, bytes]: return {state_rel: state_file.read_bytes()} def _release_outputs(self, job: JobInfo, recipients: list[str]) -> None: - """Release the outputs to ``recipients``, and push them with the job state.""" + """Release the outputs to ``recipients``, and push the job state to them. + + The sync sends the outputs, because the recipients now read them. The + data owners do not read the job state, so it goes by push. + """ job.release(["outputs"], recipients) - outputs_dir = job.job_review_path / "outputs" - files_by_datasite_path = ( - self._get_files_in_dir(outputs_dir) if outputs_dir.exists() else {} - ) - files_by_datasite_path.update(self._read_state_file(job)) - self._push_files_to_recipients(files_by_datasite_path, recipients) + self._push_files_to_recipients(self._read_state_file(job), recipients) def _push_files_to_recipients( self, files_by_datasite_path: dict, recipients: list[str] diff --git a/packages/syft-enclave/tests/test_disclosures.py b/packages/syft-enclave/tests/test_disclosures.py index d3b5c99e6f2..994be8075ed 100644 --- a/packages/syft-enclave/tests/test_disclosures.py +++ b/packages/syft-enclave/tests/test_disclosures.py @@ -373,13 +373,47 @@ def test_late_grant_reaches_data_owners(): assert (Path(do2.jobs["j"].job_review_path) / FRAMES_FILENAME).exists() -def test_released_artifact_not_sent_twice(): +def test_released_files_travel_by_grant_not_push(monkeypatch): + """The sync sends a released file to each party with read on it. + + A push of the same file would send it twice. Only the job state is pushed, + because the data owners hold no read on it. + """ + enclave, do1, do2, ds = build_quad() + submit(ds, enclave, do1, do2, OK_CODE, [LOGS]) + pushed = [] + real_push = SyftEnclaveClient._push_files_to_recipients + + def recording_push(self, files_by_datasite_path, recipients): + pushed.extend(Path(path).name for path in files_by_datasite_path) + real_push(self, files_by_datasite_path, recipients) + + monkeypatch.setattr(SyftEnclaveClient, "_push_files_to_recipients", recording_push) + run_to_completion(enclave, do1, do2, ds, [LOGS]) + + assert set(pushed) == {"state.yaml"} + review = Path(ds.jobs["j"].job_review_path) + assert (review / "stdout.txt").exists() + assert (review / "outputs" / "r.json").exists() + + +def test_released_artifact_not_granted_every_tick(monkeypatch): + """A tick with nothing left in staging does no permission work.""" enclave, do1, do2, ds = build_quad() submit(ds, enclave, do1, do2, CRASH_CODE, [FRAMES]) run_to_completion(enclave, do1, do2, ds, [FRAMES]) - job = enclave.jobs["j"] - assert enclave._push_new_disclosures(job) == [] + granted = [] + real_grant_read = JobInfo._grant_read + + def counted_grant_read(self, filenames, recipients): + granted.append(list(filenames)) + real_grant_read(self, filenames, recipients) + + monkeypatch.setattr(JobInfo, "_grant_read", counted_grant_read) + enclave.run_jobs() + + assert granted == [] def test_approve_job_rejects_non_enclave_job(): diff --git a/packages/syft-job/src/syft_job/job.py b/packages/syft-job/src/syft_job/job.py index b26689d7255..de6a576d1eb 100644 --- a/packages/syft-job/src/syft_job/job.py +++ b/packages/syft-job/src/syft_job/job.py @@ -117,7 +117,8 @@ def release(self, filenames: Iterable[str], recipients: list[str]) -> list[str]: filenames = list(filenames) moved = self._move_to_review(filenames) released = [n for n in filenames if (self.job_review_path / n).exists()] - self._grant_read(released, recipients) + if released: + self._grant_read(released, recipients) return moved def _move_to_review(self, filenames: Iterable[str]) -> list[str]: diff --git a/packages/syft-job/src/syft_job/job_runner.py b/packages/syft-job/src/syft_job/job_runner.py index 0122ac62346..400a1fe32a6 100644 --- a/packages/syft-job/src/syft_job/job_runner.py +++ b/packages/syft-job/src/syft_job/job_runner.py @@ -153,8 +153,8 @@ def _kill_process_tree(pid: int, timeout: float = 2.0) -> None: class SyftJobRunner: """Job runner that monitors and executes approved jobs. - Reads run.sh from inbox/. Outputs and state go to review/, logs and the - return code to staging/. + Reads run.sh from inbox/. State goes to review/. Outputs, logs and the + return code go to staging/, until a release moves them to review/. """ def __init__( @@ -486,8 +486,8 @@ def _execute_job( The inbox is copied into a fresh folder outside the SyftBox, and the copy must match the approved hash before anything runs, so an edit of - the submitter's inbox never runs. Outputs and state go to review/, logs - and the return code to staging/. + the submitter's inbox never runs. State goes to review/. Outputs, logs + and the return code go to staging/, until a release moves them. Args: ref: Ref of the job to execute. diff --git a/packages/syft-job/src/syft_job/job_storage.py b/packages/syft-job/src/syft_job/job_storage.py index ac7c087fc73..e7ca93ed438 100644 --- a/packages/syft-job/src/syft_job/job_storage.py +++ b/packages/syft-job/src/syft_job/job_storage.py @@ -10,6 +10,7 @@ MigrationService, ProtocolSchema, ) +from syft_perms import SyftPermContext from .config import SyftJobConfig, is_protocol_dir_name from .job_ref import JobRef, JobStateNotFoundError, RequestRef @@ -137,7 +138,17 @@ def staging_dir(self, ref: JobRef) -> Path: return self._codec_for(ref.protocol_version).staging_dir(ref) def remove_outputs(self, ref: JobRef) -> bool: - """Delete outputs/ in review/ and in staging/. Returns True if one existed.""" + """Delete outputs/ in review/ and in staging/, and the read grant on them. + + Returns True if one of the folders existed. The grant can sit in a + permission file above outputs/, so it goes on its own, and the next + release starts with no recipient. + """ + datasite = self.config.syftbox_folder / ref.datasite_email + released = self.review_dir(ref) / "outputs" + SyftPermContext(datasite=datasite).modifier.remove_all_rules( + str(released.relative_to(datasite)), is_folder=True + ) removed = False for outputs_dir in ( self.review_dir(ref) / "outputs", diff --git a/packages/syft-job/tests/test_staging.py b/packages/syft-job/tests/test_staging.py index c0851dc3fea..ae7636cebd4 100644 --- a/packages/syft-job/tests/test_staging.py +++ b/packages/syft-job/tests/test_staging.py @@ -199,12 +199,25 @@ def test_traceback_record_is_staged(tmp_path): assert (review / FRAMES_FILENAME).exists() +def test_release_with_nothing_released_skips_permissions(tmp_path, monkeypatch): + """The permission context scans the whole datasite, so build it only to grant.""" + job, _, _ = run_job(tmp_path, OK_PY, share_logs=None) + + def no_permission_work(self): + raise AssertionError("release built a permission context") + + monkeypatch.setattr(JobInfo, "_get_perm_context", no_permission_work) + + assert job.release([FRAMES_FILENAME], [DS_EMAIL]) == [] + + def test_rerun_clears_both_locations(tmp_path): """A staged log from the old run must not survive into a later release.""" job, review, staging = run_job(tmp_path, OK_PY, share_logs=False) (review / "returncode.txt").write_text("0") assert (staging / "stdout.txt").exists() - assert (review / "outputs").exists() or (staging / "outputs").exists() + assert job.release(["outputs"], [DS_EMAIL]) == ["outputs"] + assert (review / "outputs").exists() job.rerun() @@ -212,7 +225,6 @@ def test_rerun_clears_both_locations(tmp_path): assert not (staging / "stderr.txt").exists() assert not (review / "stdout.txt").exists() assert not (review / "outputs").exists() - assert not (staging / "outputs").exists() def test_rerun_clears_crash_record(tmp_path): @@ -295,6 +307,27 @@ def test_symlinked_outputs_folder_staged_as_folder(tmp_path): assert not (staging / "outputs").exists() +def test_rerun_drops_outputs_recipients_of_old_run(tmp_path): + """A recipient of the old outputs must not read the outputs of the rerun.""" + peer = "peer@test.org" + job, _, _ = run_job(tmp_path, OK_PY, share_logs=None) + # As on a datasite: the DS review folder has its own permission file, so the + # outputs grant goes there, not inside outputs/. + JobClient(config=do_config_for(tmp_path)).setup_ds_job_folder_as_do(DS_EMAIL) + job.release(["outputs"], [DS_EMAIL, peer]) + + job.rerun() + run_approved(tmp_path) + reload_job(tmp_path).release(["outputs"], [DS_EMAIL]) + + ctx = SyftPermContext(datasite=tmp_path / "SyftBox" / DO_EMAIL) + result = ctx.open( + f"app_data/job/review/{DS_EMAIL}/{PROTOCOL_DIR}/test.job/outputs/result.txt" + ) + assert result.has_read_access(DS_EMAIL) + assert not result.has_read_access(peer) + + def test_new_run_drops_recipients_and_outputs_error_of_old_run(tmp_path): """A run that blocked its outputs must not block the outputs of the next.""" job, _, staging = run_job(tmp_path, OK_PY, share_logs=None) From 94669191684f19ddf0232ac9317d66ce8b877803 Mon Sep 17 00:00:00 2001 From: Pedro Werneck Date: Fri, 30 Oct 2026 15:42:29 -0300 Subject: [PATCH 6/6] fix: remove the stale transport import from the enclave client The rebase on dev kept an import of syft.sync.transport.protocols, which dev renamed to syft.sync.connections.protocols, so syft_enclaves.client failed to import. --- packages/syft-enclave/src/syft_enclaves/client.py | 1 - 1 file changed, 1 deletion(-) diff --git a/packages/syft-enclave/src/syft_enclaves/client.py b/packages/syft-enclave/src/syft_enclaves/client.py index b5599eca51e..1f2b6d7c29e 100644 --- a/packages/syft-enclave/src/syft_enclaves/client.py +++ b/packages/syft-enclave/src/syft_enclaves/client.py @@ -4,7 +4,6 @@ from pathlib import Path from typing import TYPE_CHECKING, Optional -from syft.sync.transport.protocols import Transport from syft_datasets import Dataset from syft_datasets.dataset_manager import SyftDatasetManager from syft_job.disclosures import DisclosuresArg, gated_names