Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
17 commits
Select commit Hold shift + click to select a range
0ebb07e
fix: make silent job and peer failures visible and actionable
pjwerneck Sep 1, 2026
288ec39
Merge remote-tracking branch 'origin/dev' into pjwerneck/fix-silent-f…
koenvanderveen Sep 2, 2026
6b53ac6
Merge remote-tracking branch 'origin/dev' into pjwerneck/fix-silent-f…
koenvanderveen Sep 2, 2026
e6398dd
Merge pull request #9501 from OpenMined/koen/pr-review-skill-standards
koenvanderveen Sep 2, 2026
b964e32
Merge pull request #9502 from OpenMined/koen/pr-review-skill-standards
koenvanderveen Sep 2, 2026
bd1ebb0
Merge pull request #9504 from OpenMined/koen/pr-review-skill-naming
koenvanderveen Sep 2, 2026
f316fac
Merge pull request #9505 from OpenMined/koen/pr-review-skill-naming
koenvanderveen Sep 2, 2026
8fd723f
fix: update job indexing to use datasite and job name
pjwerneck Sep 3, 2026
7a4af9b
fix: update job referencing to use email instead of datasite for clarity
pjwerneck Sep 3, 2026
d29a2fd
fix: revert wrong pre-check, move tests
pjwerneck Sep 3, 2026
8d310be
test: split job addressing tests for hinting and indexing
pjwerneck Sep 3, 2026
4457bb1
fix: prettier pre-commit fail
pjwerneck Sep 3, 2026
f65467e
chore: number every review section and give each PR its own review fo…
koenvanderveen Sep 4, 2026
3221c65
Merge branch 'dev' into pjwerneck/fix-silent-failures
pjwerneck Sep 8, 2026
cf3b552
Merge branch 'dev' into pjwerneck/fix-silent-failures
pjwerneck Sep 18, 2026
71e05ca
Merge remote-tracking branch 'origin/dev' into pjwerneck/fix-silent-f…
pjwerneck Oct 7, 2026
727dc7d
fix: stop duplicate package logs and give ambiguous job names working…
pjwerneck Oct 7, 2026
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -100,14 +100,15 @@ Submit the job and retrieve results:
ds.submit_python_job(
user="do@org.com",
code_path="analysis.py",
job_name="analysis",
)
ds.sync(); do.sync()

# Data owner Approves & runs job
do.jobs[0].approve()
do.jobs["ds@org.com"]["analysis"].approve()
do.process_approved_jobs(share_outputs_with_submitter=True)
do.sync(); ds.sync()
result = open(ds.jobs[-1].output_paths[0]).read()
result = open(ds.jobs["do@org.com"]["analysis"].output_paths[0]).read()
```

## Packages
Expand Down
51 changes: 50 additions & 1 deletion docs/API.md
Original file line number Diff line number Diff line change
Expand Up @@ -59,7 +59,43 @@ Returns a `PeerList`.

Get the list of jobs. Auto-syncs before returning.

Returns a `JobsList`.
Returns a `JobsList`. Address a job by **email, then name** — a job name is
unique per datasite and submitter, so the email is what makes the name resolve
to one job, and both parts stay the same as jobs are added:

```python
# as the data scientist, naming the data owner
client.jobs["do@org.com"]["analysis"].output_paths

# as the data owner, naming the submitter
client.jobs["ds@org.com"]["analysis"].approve()
```

**An email keeps the jobs it is a party to**, on either side: the datasite they
sit on, or the person who submitted them. Usually that is the other party — a
data scientist names the data owner, a data owner names the submitter — but
naming yourself works and keeps your own, which is what a `PermissionError` on
someone else's job suggests. Job names cannot contain `@`, so the two kinds of
key never collide.

Chain both emails when one submitter sent the same name to two datasites:

```python
client.jobs["do@org.com"]["ds@org.com"]["analysis"].approve()
```

A bare name (`client.jobs["analysis"]`) searches every datasite at once. It
still works, but it raises when more than one job answers to it, and the message
names whichever key separates them.

A job name can no longer hold an `@`. A job submitted before that rule still
resolves by name, with a `DeprecationWarning`; the next version will not resolve
it, so rename such a job.

Positional indexing (`client.jobs[0]`) also works and matches the Index column
in the table, but positions shift as jobs are added, so prefer an email and a
name. A position is worth using in one case: two jobs that share a datasite, a
submitter and a name, which no email separates.

### `client.datasets`

Expand Down Expand Up @@ -174,6 +210,7 @@ Submit a Python job to a Data Owner. **DS only.**
ds_client.submit_python_job(
user="owner@example.com",
code_path="/path/to/script.py",
job_name="analysis",
)
```

Expand All @@ -195,11 +232,23 @@ Run all approved jobs. **DO only.**
- `stream_output`: Stream stdout/stderr in real-time.
- `timeout`: Timeout in seconds per job (default: 300).
- `force_execution`: Skip version compatibility checks.
- `ignore_peer_version`: Run jobs from peers whose version is incompatible.

```python
do_client.process_approved_jobs()
```

A job whose submitter runs an incompatible version is not run. Each one is
reported by name, with the submitter and the reason:

```
⏭️ 1 approved job(s) did not run:
• analysis (submitted by ds@example.com): Skipping peer ds@example.com: incompatible version.
Pass ignore_peer_version=True to run them anyway.
```

The job stays at `approved`, so it runs on the next call once the versions match.

---

## Cleanup
Expand Down
6 changes: 6 additions & 0 deletions packages/syft-bg/src/syft_bg/__init__.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,7 @@
__version__ = "0.2.2"

from syft_job.logging_config import configure_package_logger

from syft_bg.api import (
AuthResult,
AutoApproveResult,
Expand Down Expand Up @@ -61,3 +63,7 @@ def __getattr__(name: str):
print("No config file found, run syft_bg.init() first")
return
raise AttributeError(f"module 'syft_bg' has no attribute {name!r}")


# Last, so the imports stay at the top. Nothing here logs at import time.
configure_package_logger(__name__)
2 changes: 2 additions & 0 deletions packages/syft-enclave/pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@ requires-python = ">=3.10"
dependencies = [
"syft>=0.10.0", # floor: the `syft` PyPI project also hosts legacy PySyft <=0.9
"syft-rds>=0.6.1",
"syft-job==0.1.40", # imported directly for logging_config
"pydantic-settings>=2.11.0",
"requests>=2.32.0",
"google-auth[pyjwt]>=2.22.0",
Expand All @@ -32,6 +33,7 @@ build-backend = "hatchling.build"
[tool.uv.sources]
"syft" = { workspace = true }
"syft-rds" = { workspace = true }
"syft-job" = { workspace = true }

[tool.hatch.build.targets.wheel]
packages = ["src/syft_enclaves"]
Expand Down
4 changes: 4 additions & 0 deletions packages/syft-enclave/src/syft_enclaves/__init__.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
from syft_job.logging_config import configure_package_logger
from syft_enclaves.client import SyftEnclaveClient
from syft_enclaves.login import login_do, login_ds
from syft_enclaves.runner import EnclaveRunner
Expand All @@ -10,3 +11,6 @@
"login_do",
"login_ds",
]

# Last, so the imports stay at the top. Nothing here logs at import time.
configure_package_logger(__name__)
2 changes: 2 additions & 0 deletions packages/syft-enclave/src/syft_enclaves/__main__.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@
import sys

from pydantic import ValidationError
from syft_job.logging_config import defer_to_root_logging

from syft_enclaves.client import SyftEnclaveClient
from syft_enclaves.receipt.writer import ReceiptSettings
Expand All @@ -23,6 +24,7 @@ def _configure_logging(log_level: str) -> None:
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)
defer_to_root_logging()


def _load_settings() -> EnclaveSettings:
Expand Down
2 changes: 2 additions & 0 deletions packages/syft-enclave/src/syft_enclaves/bootstrap.py
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@
from pathlib import Path

import requests
from syft_job.logging_config import defer_to_root_logging

from syft_enclaves._unix_socket import UnixSocketConnection
from syft_enclaves.evidence.tinfoil import TINFOIL_ATTESTATION_PATH
Expand Down Expand Up @@ -205,6 +206,7 @@ def run() -> None:
level="INFO",
format="%(asctime)s [%(levelname)s] bootstrap: %(message)s",
)
defer_to_root_logging()

token_path = Path(
os.environ.get("SYFT_ENCLAVE_TOKEN_PATH", str(DEFAULT_TOKEN_PATH))
Expand Down
2 changes: 1 addition & 1 deletion packages/syft-enclave/src/syft_enclaves/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -322,7 +322,7 @@ def _wrap_enclave_jobs(self, jobs_list: JobsList) -> JobsList:
self._as_enclave_job(j) if j.job_headers.get("job_type") == "enclave" else j
for j in jobs_list
]
return JobsList(wrapped, jobs_list._root_email)
return JobsList(wrapped, jobs_list._root_email, jobs_list._has_do_role)

def wait_until_peered(
self,
Expand Down
7 changes: 5 additions & 2 deletions packages/syft-enclave/src/syft_enclaves/enclave_job_info.py
Original file line number Diff line number Diff line change
Expand Up @@ -117,8 +117,11 @@ def _approval_file(self) -> Path:
def _load_own_approval(self) -> PartyApprovalStatus:
if not self._approval_file.exists():
raise PermissionError(
f"No approval file found for {self.current_user_email}. "
f"You may not be a designated party for this job."
f"No approval file for {self.current_user_email} on job "
f"'{self.name}'. The enclave writes one per designated party "
f"when it distributes the job, so either it has not distributed "
f"this job yet — run client.sync() and retry — or you are not a "
f"party to it."
)
return PartyApprovalStatus.load_json(self._approval_file)

Expand Down
82 changes: 82 additions & 0 deletions packages/syft-enclave/tests/test_enclave_job_info.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,82 @@
"""Unit tests for EnclaveJobInfo, the per-party approval gate.

The gate lives here rather than in SyftEnclaveClient.approve_job, so these
build a job on a tmp_path SyftBox folder instead of a four-party enclave flow.
"""

from datetime import datetime, timezone
from pathlib import Path

import pytest
from syft_enclaves.enclave_job_info import (
EnclaveJobInfo,
PartyApprovalStatus,
enclave_approval_file_name,
)
from syft_job.client import JobClient
from syft_job.config import SyftJobConfig
from syft_job.job import JobInfo
from syft_job.job_storage import JobRef
from syft_job.models import JobState, JobStatus, JobSubmissionMetadata

DO_EMAIL = "do@test.org"
DS_EMAIL = "ds@test.org"


def _make_enclave_job(tmp_path: Path, job_name: str = "test_job") -> EnclaveJobInfo:
"""An enclave job on the DO's datasite, with no approval file written yet."""
syftbox = tmp_path / "SyftBox"
syftbox.mkdir()
client = JobClient(
config=SyftJobConfig(syftbox_folder=syftbox, current_user_email=DO_EMAIL)
)
ref = JobRef(
datasite_email=DO_EMAIL,
ds_email=DS_EMAIL,
job_name=job_name,
protocol_version="1",
)
job = JobInfo(
job_metadata=JobSubmissionMetadata(
name=job_name,
type="python",
submitted_by=DS_EMAIL,
datasite_email=DO_EMAIL,
submitted_at=datetime.now(timezone.utc),
),
state=JobState(status=JobStatus.PENDING),
client=client,
current_user_email=DO_EMAIL,
ref=ref,
)
return EnclaveJobInfo.from_job_info(job)


def test_approve_refuses_when_approval_file_missing(tmp_path: Path):
"""No approval file means the enclave has not distributed the job yet.

The message used to say the caller may not be a designated party, which is
the wrong cause for the common case and offers nothing to do about it.
"""
job = _make_enclave_job(tmp_path)

with pytest.raises(PermissionError) as exc:
job.approve()

message = str(exc.value)
assert DO_EMAIL in message
assert "test_job" in message
assert "client.sync()" in message


def test_approve_refuses_when_already_approved(tmp_path: Path):
"""A second approval must not overwrite the first one's timestamp."""
job = _make_enclave_job(tmp_path)
approval_file = job.job_review_path / enclave_approval_file_name(DO_EMAIL)
PartyApprovalStatus(party=DO_EMAIL).save_json(approval_file)

job.approve()
assert PartyApprovalStatus.load_json(approval_file).status == JobStatus.APPROVED

with pytest.raises(ValueError, match="Already in status: approved"):
job.approve()
4 changes: 4 additions & 0 deletions packages/syft-job/src/syft_job/__init__.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
# __version__ comes from the installed distribution metadata (see version.py).
from .version import __version__
from .logging_config import configure_package_logger

from .client import BaseJobClient, JobClient, get_client
from .config import SyftJobConfig
Expand Down Expand Up @@ -32,3 +33,6 @@
# Migration registry
"job_registry",
]

# Last, so the imports stay at the top. Nothing here logs at import time.
configure_package_logger(__name__)
27 changes: 15 additions & 12 deletions packages/syft-job/src/syft_job/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -741,8 +741,18 @@ def jobs(self) -> JobsList:
"""
Get all jobs from all peer directories as an indexable list grouped by user.

Order is the same as the jobs table: jobs on your own datasite first,
then the other datasites alphabetically, newest-first within each one.
``jobs[N]`` is the row labelled ``[N]``. Prefer an email and a name
(``jobs["do@org.com"]["analysis"]``); both parts stay the same as jobs
are added, and the email is what makes the name resolve to one job.

Returns a JobsList object that can be:
- Indexed: jobs[0], jobs[1], etc.
- Indexed by email, then name (preferred): jobs["do@org.com"]["analysis"]
- Indexed by datasite and submitter, for a name they share:
jobs["do@org.com"]["ds@org.com"]["analysis"]
- Indexed by name across every datasite: jobs["analysis"]
- Indexed by position: jobs[0], jobs[1] — matches the table Index column
- Iterated: for job in jobs
- Displayed: print(jobs) shows separate tables for each user
- HTML display: in Jupyter, shows separate tables for each user with jobs
Expand All @@ -755,25 +765,18 @@ def jobs(self) -> JobsList:

current_jobs = self._get_all_jobs()

# Sort jobs by recent submissions first (newest first), then by user/status
def job_sort_key(job):
# Parse submitted_at timestamp for sorting (most recent first)
# Root owner first, then peers, newest-first within each owner.
# This list is the authority for both jobs[N] and the table's [N].
Comment thread
pjwerneck marked this conversation as resolved.
try:
if job.submitted_at:
from datetime import datetime as dt

# Parse ISO format timestamp
ts = dt.fromisoformat(job.submitted_at.replace("Z", "+00:00"))
# Use negative timestamp for reverse chronological order (newest first)
ts = datetime.fromisoformat(job.submitted_at.replace("Z", "+00:00"))
time_priority = -ts.timestamp()
else:
# Jobs without submitted_at go to the end
time_priority = float("inf")
except Exception:
# Invalid timestamps go to the end
time_priority = float("inf")

# Secondary sorting: user priority (root first), then user name, then status
user_priority = (
0 if job.datasite_owner_email == self.current_user_email else 1
)
Expand All @@ -789,9 +792,9 @@ def job_sort_key(job):
status_priority = status_order.get(job.status, 7)

return (
time_priority,
user_priority,
job.datasite_owner_email,
time_priority,
status_priority,
job.name.lower(),
)
Expand Down
Loading
Loading