feat: release job outputs and logs through one path

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
 

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."
    ]
   },
   {

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

@@ -1,4 +1,3 @@
-import json
 import logging
 import os
 from datetime import datetime, timezone
@@ -50,7 +49,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"
@@ -474,30 +472,38 @@ 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. 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_artifacts(gated_names(self.granted_disclosures(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."""
+        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."""
         for job in self.jobs:
             if job.status not in ("done", "failed"):
                 continue
-            self._share_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._release_results_once(job)
 
         self._rds.sync()
 
-    def _share_results_once(self, job: JobInfo) -> None:
-        """Send the outputs and the state, the first time the job finishes."""
+    def _release_results_once(self, job: JobInfo) -> None:
+        """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
@@ -506,55 +512,16 @@ def _share_results_once(self, job: JobInfo) -> None:
             self._try_write_receipt(job)
 
         # Always share results with the DS (submitter)
-        self._forward_results_to_recipients(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:
-            do_emails = list(datasets.keys())
-            job.share_outputs(do_emails)
-            self._forward_results_to_recipients(job, do_emails)
+            do_emails = list(datasets)
+            self._release_outputs(job, do_emails)
 
         marker.write_text("shared")
 
-    def _forward_new_releases(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
-        }
-        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)
-
-        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.
 
@@ -576,14 +543,14 @@ 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."""
-        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)
+    def _release_outputs(self, job: JobInfo, recipients: list[str]) -> None:
+        """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)
+        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]

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"))

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])
@@ -360,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._forward_new_releases(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():

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

@@ -97,28 +97,35 @@ 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], recipients: list[str]) -> list[str]:
+        """Move the named staged files to review/, for ``recipients`` to read.
 
-        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. 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.
         """
-        moved = self.release_artifacts(filenames)
-        self._grant_read(moved, [self.submitted_by])
+        if not recipients:
+            return []
+        filenames = list(filenames)
+        moved = self._move_to_review(filenames)
+        released = [n for n in filenames if (self.job_review_path / n).exists()]
+        if released:
+            self._grant_read(released, recipients)
         return moved
 
-    def release_artifacts(self, filenames: Iterable[str]) -> list[str]:
-        """Move the named staged artifacts into the review directory.
+    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 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:
@@ -204,8 +211,11 @@ 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."""
-        return self._release_to_submitter(gated_names(self.granted_disclosures))
+        """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])
 
     # ──────────────────────────────────────────────
     # Properties from config (inbox/)
@@ -265,7 +275,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 +543,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 +558,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,9 +618,7 @@ 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)
+        if self._client.manager.remove_outputs(self._ref):
             changes_made.append("outputs directory")
 
         # Reset state to approved
@@ -643,25 +651,13 @@ 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."""
+    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