feat: add per-party disclosure for enclave job logs and traceback frames

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

@@ -69,7 +69,7 @@ def _read_job_stderr(
         return None, None
 
     stderr_text = None
-    stderr_file = job.job_review_path / "stderr.txt"
+    stderr_file = job.artifact_path("stderr.txt")
     if stderr_file.exists():
         try:
             file_size = stderr_file.stat().st_size

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

@@ -1,6 +1,7 @@
 from datetime import datetime, timezone
 from pathlib import Path
-from typing import Optional
+from typing import Iterable, Optional
+import json
 import os
 
 from syft_rds import SyftRDSClient, SyftRDSClientConfig
@@ -12,9 +13,13 @@
 from syft_job.job_storage import JobRef
 from syft_job.models import JobState, JobStatus
 
+from syft_job.traceback_capture import FRAMES_FILENAME
+
 from syft_enclaves.enclave_job_info import (
+    DisclosureItem,
     EnclaveJobInfo,
     PartyApprovalStatus,
+    approved_disclosures,
     enclave_approval_file_name,
 )
 from syft_enclaves.attestation import (
@@ -37,6 +42,32 @@
 )
 
 
+# The artifacts that move only under the agreement of every party. Everything
+# else in the review folder stays readable by the submitter.
+FORWARDED_RECORD = "forwarded_artifacts.json"
+RESULTS_SHARED_MARKER = "results_shared"
+
+GATED_ARTIFACTS = {
+    DisclosureItem.LOGS.value: ("stdout.txt", "stderr.txt"),
+    DisclosureItem.TRACEBACK_FRAMES.value: (FRAMES_FILENAME,),
+}
+
+
+def gated_names(granted: Iterable[str]) -> list[str]:
+    """The filenames that the granted disclosure items cover."""
+    return [
+        name
+        for item, names in GATED_ARTIFACTS.items()
+        if item in granted
+        for name in names
+    ]
+
+
+def pre_sync_enabled() -> bool:
+    """Whether the client syncs on its own. ``PRE_SYNC=false`` turns it off."""
+    return os.environ.get("PRE_SYNC", "true").lower() == "true"
+
+
 class SyftEnclaveClient:
     def __init__(
         self,
@@ -162,6 +193,7 @@ def submit_python_job(
         job_name: Optional[str] = "",
         datasets: Optional[dict[str, list[str]]] = None,
         share_results_with_do: bool = False,
+        request_disclosures: Optional[list[str]] = None,
         force_submission: bool = False,
         ignore_peer_version: bool = False,
         **kwargs,
@@ -187,6 +219,7 @@ def submit_python_job(
             job_name,
             datasets=datasets,
             share_results_with_do=share_results_with_do,
+            request_disclosures=request_disclosures,
             **kwargs,
         )
         self._rds.sync_engine.push_job_files(job_dir)
@@ -203,35 +236,112 @@ def run_jobs(self) -> None:
                     state.status = JobStatus.APPROVED
                     state.save(job.job_review_path / "state.yaml")
 
+        # share_logs_with_submitter=False keeps the logs in staging. Only a
+        # grant from every party releases them.
         self._rds.process_approved_jobs(
             force_execution=True,
             share_outputs_with_submitter=True,
-            share_logs_with_submitter=True,
+            share_logs_with_submitter=False,
+        )
+        self._apply_disclosure_policy()
+
+    def _requested_disclosures(self, job: JobInfo) -> list[str]:
+        """The items the submitter asked for, as recorded at submission."""
+        requested = job.job_headers.get("requested_disclosures")
+        return list(requested) if isinstance(requested, list) else []
+
+    def granted_disclosures(self, job: JobInfo) -> set[str]:
+        """The items that every party released for this job."""
+        return approved_disclosures(
+            job.job_review_path, self._requested_disclosures(job)
         )
 
+    def _local_jobs(self) -> JobsList:
+        """The job list read from disk, with no sync.
+
+        ``self.jobs`` syncs first, therefore a caller that must act before a
+        push reads the list through this method.
+        """
+        return self._rds.job_client.jobs
+
+    def _apply_disclosure_policy(self) -> None:
+        """Move each granted artifact from the staging folder into review.
+
+        An artifact that no party released stays in the staging folder, where no
+        submitter 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)))
+
     def distribute_results(self) -> None:
         """Distribute job results to DS (always) and optionally to DOs."""
         for job in self.jobs:
-            if job.status != "done":
-                continue
-            results_shared_marker = job.job_review_path / "results_shared"
-            if results_shared_marker.exists():
+            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)
 
-            # Always share results with the DS (submitter)
-            self._forward_results_to_recipients(job, [job.submitted_by])
+        self._rds.sync()
 
-            # Optionally share with DOs
-            if job.job_headers.get("share_results_with_do"):
-                datasets = job.job_metadata.datasets
-                if datasets:
-                    do_emails = list(datasets.keys())
-                    job.share_outputs(do_emails)
-                    self._forward_results_to_recipients(job, do_emails)
+    def _share_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():
+            return
 
-            results_shared_marker.write_text("shared")
+        # Always share results with the DS (submitter)
+        self._forward_results_to_recipients(job, [job.submitted_by])
 
-        self._rds.sync()
+        # 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)
+
+        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 _read_state_file(self, job: JobInfo) -> dict[Path, bytes]:
         """Read the job state.yaml as a {path_in_datasite: bytes} dict."""
@@ -245,10 +355,16 @@ def _read_state_file(self, job: JobInfo) -> dict[Path, 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"
-        if not outputs_dir.exists():
-            return
-        files_by_datasite_path = self._get_files_in_dir(outputs_dir)
+        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 _push_files_to_recipients(
+        self, files_by_datasite_path: dict, recipients: list[str]
+    ) -> None:
+        """Queue the files for the recipients through the event outbox."""
         if not files_by_datasite_path:
             return
         syncer = self._rds.sync_engine.datasite_owner_syncer
@@ -261,12 +377,24 @@ def _forward_results_to_recipients(self, job: JobInfo, recipients: list[str]):
         )
         syncer.process_syftbox_events_queue()
 
-    def approve_job(self, job: JobInfo) -> None:
-        """Approve an enclave job and push the approval state file to the enclave."""
-        if os.environ.get("PRE_SYNC", "true").lower() == "true":
+    def approve_job(
+        self, job: JobInfo, disclosures: Optional[Iterable[str]] = None
+    ) -> None:
+        """Approve an enclave job and push the approval state file to the enclave.
+
+        ``disclosures`` names the items in ``DisclosureItem`` that this data
+        owner releases. The enclave releases an item only when every data owner
+        released it. Omit the argument to release nothing.
+        """
+        if not isinstance(job, EnclaveJobInfo):
+            raise TypeError(
+                f"Job '{job.name}' is not an enclave job, so it carries no "
+                f"disclosures. Approve it through the datasite client."
+            )
+        if pre_sync_enabled():
             self._rds.sync()
 
-        job.approve()
+        job.approve(disclosures=disclosures)
         file_name = enclave_approval_file_name(self.email)
         approval_file = job.job_review_path / file_name
         if not approval_file.exists():
@@ -278,6 +406,24 @@ def approve_job(self, job: JobInfo) -> None:
             relative_path, process_now=True
         )
 
+    def update_disclosures(
+        self, job: JobInfo, disclosures: Optional[Iterable[str]]
+    ) -> dict:
+        """Change the items this data owner releases, and push the new set.
+
+        The enclave applies the new set on its next cycle.
+        """
+        if pre_sync_enabled():
+            self._rds.sync()
+
+        result = job.update_disclosures(disclosures)
+        approval_file = job.job_review_path / enclave_approval_file_name(self.email)
+        relative_path = approval_file.relative_to(self._rds.syftbox_folder)
+        self._rds.sync_engine.datasite_watcher_syncer.on_file_change(
+            relative_path, process_now=True
+        )
+        return result
+
     def receive_jobs(self):
         """Receive and distribute enclave jobs to relevant DOs.
 

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

@@ -5,6 +5,8 @@
 from syft_job.job import JobsList
 from syft_job.models import JobSubmissionMetadata
 
+from syft_enclaves.enclave_job_info import normalize_disclosures
+
 
 class EnclaveJobClient(BaseJobClient):
     """Wraps a JobClient to add enclave-specific job submission behavior.
@@ -52,6 +54,7 @@ def submit_python_job(
         job_name: Optional[str] = "",
         datasets: Optional[dict[str, list[str]]] = None,
         share_results_with_do: bool = False,
+        request_disclosures: Optional[list[str]] = None,
         **kwargs,
     ) -> Path:
         """Submit a Python job with enclave metadata.
@@ -69,6 +72,9 @@ def submit_python_job(
         config.headers = {
             "job_type": "enclave",
             "share_results_with_do": share_results_with_do,
+            # The items the submitter asks for. Each data owner sees this list
+            # next to the code, then releases none, some, or all of it.
+            "requested_disclosures": sorted(normalize_disclosures(request_disclosures)),
         }
         config.save(job_dir / "config.yaml")