docs: describe policies in plain words + add a policies step to the DO tutorial

docs/jobs.md

@@ -171,9 +171,27 @@ The runner gives each policy these environment variables:
 
 The module docstring of `syft_job/policies/env.py` documents every variable and convention in one place.
 
+### Describing a policy
+
+The DS sees the full source of an API's policies when they look at the API, for example in `ds_client.api["word_count"]`. Start every policy with a docstring of one or two lines that says in plain words what the policy allows and what it refuses, so that is the first thing the DS reads:
+
+```python
+"""Allows each caller up to 10 calls every 24 hours, and refuses any call over that limit."""
+```
+
+A bash policy can start with a comment instead.
+
 ### Denying a call
 
-A policy denies a call by exiting with a non-zero code. In Python, any uncaught exception does that. The job is then marked `failed`. The last line of the policy's error output is written to `policy_error.txt` and shared with the DS. The full error output stays with the DO, because a traceback can show private paths or data.
+A policy denies a call by raising an exception, or by exiting with a non-zero code. The job is then marked `failed`. The last line of the policy's error output is written to `policy_error.txt` and shared with the DS, who reads it in `job.policy_error`. The full error output stays with the DO, because a traceback can show private paths or data.
+
+For an uncaught exception, that last line is the exception's name and message. Keep the message to one line that says what was refused and why:
+
+```python
+raise ValueError("Rate limit reached: 10 calls per 24 hours")
+```
+
+The DS then sees `ValueError: Rate limit reached: 10 calls per 24 hours`.
 
 ### State and locking
 
@@ -182,6 +200,8 @@ The state folder belongs to the API, and only the DO can read it. The policy wri
 ### Example: rate limit
 
 ```python
+"""Allows each caller up to 10 calls every 24 hours, and refuses any call over that limit."""
+
 import json
 import os
 from datetime import datetime, timedelta, timezone
@@ -246,21 +266,27 @@ The DS always sees the job's status. When the output policy denies, the job is `
 ### Example: block a word
 
 ```python
+"""Shares the outputs only when no output file contains the word SECRET.
+Refuses to share any of them when one does."""
+
 import os
 from pathlib import Path
 
 outputs = Path(os.environ["SYFT_POLICY_OUTPUTS_DIR"])
 
 for f in outputs.rglob("*"):
     if f.is_file() and "SECRET" in f.read_text(errors="ignore"):
-        raise ValueError(f"Output {f.name} contains a blocked word")
+        raise ValueError(f"Outputs not shared: {f.name} contains a blocked word")
 ```
 
 ### Example: redact CSV rows
 
 Only `outputs/` is shared, so a job and its output policy can agree on another folder for staging. In this example the job writes `results.csv` to `output_candidates/`, next to `outputs/`. The policy replaces every row that mentions the secret, writes the cleaned file to `outputs/`, and logs the original rows in the state folder. `output_candidates/` is a convention of this example; the runner does not know about it.
 
 ```python
+"""Shares results.csv with every row that mentions SECRET replaced by [REDACTED].
+The original rows stay with the data owner."""
+
 import csv
 import os
 from pathlib import Path
@@ -295,6 +321,8 @@ result.outputs_dir   # the outputs after the run
 ### Example: add an auditor
 
 ```python
+"""Shares the outputs with the caller and with auditor@example.com."""
+
 import json
 import os
 from pathlib import Path

notebooks/ai_audit/external/DS_Tutorial_v2.ipynb

@@ -1048,7 +1048,9 @@
    "source": [
     "### 11d — Call the api\n",
     "\n",
-    "Once the DO has created an api from the job (they name it after the job by default), you can call it. The call sends a request with only your parameters, no code. The argument names come from `params.json`. With `block=False` the call returns the request right away (`call.job` is the job once the DO accepts it); by default it waits and returns the finished job."
+    "Once the DO has created an api from the job (they name it after the job by default), you can call it. The call sends a request with only your parameters, no code. The argument names come from `params.json`. With `block=False` the call returns the request right away (`call.job` is the job once the DO accepts it); by default it waits and returns the finished job.\n",
+    "\n",
+    "The DO can also attach policies to the api: one can refuse your call before it runs, or hold its outputs back afterwards. If a policy refuses, the job is `failed` and the DO's reason is in `job.policy_error`."
    ]
   },
   {

notebooks/ai_audit/internal/DO_Tutorial_V3.ipynb

@@ -35,7 +35,7 @@
     "| 8 | Review and approve (or reject) a job |\n",
     "| 9 | Run approved jobs and share results |\n",
     "| 10 | Run unattended with `syft_bg` background services |\n",
-    "| 11 | Turn a reviewed job into an api |\n",
+    "| 11 | Turn a reviewed job into an api, and add a policy |\n",
     "| 12 | Inspect service logs |\n",
     "| 13 | Reset (only if needed) |\n",
     "\n",
@@ -46,7 +46,7 @@
     "2. **OAuth client credentials** (`credentials.json` from Google Cloud Console). One-time setup; see Step 3.\n",
     "3. **The Data Scientist's email** — they'll send you a peer request, and you decide whether to approve it.\n",
     "\n",
-    "> 💡 This is a **Jupyter notebook running on your local machine** (laptop, desktop, or always-on server). Data Owners need a persistent host because the background services in Step 11 keep running between sessions. If you're running Jupyter on a *headless* remote server (no browser/display), see the heads-up in Step 3 about the OAuth flow.\n"
+    "> 💡 This is a **Jupyter notebook running on your local machine** (laptop, desktop, or always-on server). Data Owners need a persistent host because the background services in Step 10 keep running between sessions. If you're running Jupyter on a *headless* remote server (no browser/display), see the heads-up in Step 3 about the OAuth flow.\n"
    ]
   },
   {
@@ -825,7 +825,7 @@
     "|---------|--------------|\n",
     "| 🔄 **`sync`** | Keeps your SyftBox in sync with Drive — new submissions appear without manual `client.sync()` calls. |\n",
     "| 🔔 **`notify`** | Sends you an email when a new job arrives or a peer requests connection. |\n",
-    "| 🤖 **`approve`** | Approves and runs calls of your apis (Step 12) — and, optionally, auto-approves peer requests from whitelisted domains. |\n",
+    "| 🤖 **`approve`** | Approves and runs calls of your apis (Step 11) — and, optionally, auto-approves peer requests from whitelisted domains. |\n",
     "| 📧 **`email_approve`** | Lets you approve/reject jobs by replying to the notification emails. Requires a Google Cloud Pub/Sub topic. |\n",
     "\n",
     "### Three-stage setup\n",
@@ -923,9 +923,9 @@
    },
    "source": [
     "---\n",
-    "## Step 10 — Apis for trusted job patterns\n",
+    "## Step 11 — Apis for trusted job patterns\n",
     "\n",
-    "Reviewing every job by hand gets old fast for repeat workflows. An **api** is a job you approved in advance: the DS *calls* it with their own parameters, and each call runs without you reviewing it again. The api pins the code (`run.sh` and the Python file under `code/`, matched by content hash), and takes `config.yaml` and `code/params.json` by name, so the DS can change the parameters between calls.\n",
+    "Reviewing every job by hand gets old fast for repeat workflows. An **api** is a job you approved in advance: the DS *calls* it with their own parameters, and each call runs without you reviewing it again. The api pins the code (`run.sh` and the Python file under `code/`, matched by content hash), and takes `code/params.json` by name, so the DS can change the parameters between calls.\n",
     "\n",
     "> 💡 Calls of your apis need the `approve` service from Step 10 to run on their own. Without it, call `client.process_approved_jobs(...)` yourself.\n",
     "\n",
@@ -934,7 +934,9 @@
     "| Match type | When to use it |\n",
     "|------------|----------------|\n",
     "| **By content (hash)** | The file shouldn't change. Used for `run.sh` and `main.py` — a call can't change the code. |\n",
-    "| **By name only** | The file's name is fixed but its contents change between calls. Used for `config.yaml` and `code/params.json`. |\n",
+    "| **By name only** | The file's name is fixed but its contents change between calls. Used for `code/params.json`. |\n",
+    "\n",
+    "`config.yaml` is matched neither way: the api leaves it out, because your client writes it for each call.\n",
     "\n",
     "### The cleanest pattern\n",
     "\n",
@@ -943,8 +945,8 @@
     "```python\n",
     "syft_bg.create_api_from_job(\n",
     "    job,\n",
-    "    # code/params.json and config.yaml are name-matched by default;\n",
-    "    # all other files are content-matched\n",
+    "    # code/params.json is name-matched by default, config.yaml is left out\n",
+    "    # (your client writes it for each call), all other files are content-matched\n",
     "    # peers=[ds_email]            # optional: restrict to a specific DS\n",
     ")\n",
     "```\n",
@@ -986,7 +988,8 @@
    },
    "outputs": [],
    "source": [
-    "# Create the api. run.sh and main.py are content-hashed; config.yaml and code/params.json match by name.\n",
+    "# Create the api. run.sh and main.py are content-hashed; code/params.json matches by name.\n",
+    "# config.yaml is left out: your client writes it for each call.\n",
     "# The api is named after the job. Pass name=\"...\" to choose another name.\n",
     "\n",
     "result = syft_bg.create_api_from_job(candidate)\n",
@@ -1000,6 +1003,124 @@
   {
    "cell_type": "markdown",
    "id": "45",
+   "metadata": {
+    "id": "api-policy-md"
+   },
+   "source": [
+    "### Add an input policy\n",
+    "\n",
+    "A **policy** is a short script that you attach to an api. An **input policy** runs before each call's job and decides whether the call may run. An **output policy** runs after the job and checks the outputs before they're shared ([`docs/jobs.md`](https://github.com/OpenMined/PySyft/blob/dev/docs/jobs.md#output-policies) has examples).\n",
+    "\n",
+    "The input policy below allows calls on your ocean dataset only, up to 3 per DS every 24 hours. Two conventions make it easy for the DS to read:\n",
+    "\n",
+    "- **Start with a one- or two-line docstring** that says in plain words what the policy allows and refuses. The DS sees the policy's full source when they look at your api in `client.api`, so this is the first thing they read.\n",
+    "- **Refuse with `raise ValueError(\"<message>\")`.** The DS sees the exception's name and message in `job.policy_error`, as in `ValueError: Rate limit reached: 3 calls per 24 hours`, so keep the message to one line that says what was refused and why.\n",
+    "\n",
+    "Set `ALLOWED_DATASETS` to the dataset name from Step 6a. If you change it or `MAX_CALLS`, change the docstring too."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "46",
+   "metadata": {
+    "id": "api-policy-write-cell"
+   },
+   "outputs": [],
+   "source": [
+    "%%writefile input_policy.py\n",
+    "\"\"\"Allows calls on the ocean_water_quality2 dataset only, up to 3 per person every 24 hours.\n",
+    "Refuses calls on any other dataset, and any call over that limit.\"\"\"\n",
+    "\n",
+    "import json\n",
+    "import os\n",
+    "from datetime import datetime, timedelta, timezone\n",
+    "from pathlib import Path\n",
+    "\n",
+    "ALLOWED_DATASETS = {\"ocean_water_quality2\"}\n",
+    "MAX_CALLS = 3\n",
+    "\n",
+    "dataset = json.loads(Path(os.environ[\"SYFT_POLICY_PARAMS_PATH\"]).read_text()).get(\"dataset\")\n",
+    "if dataset not in ALLOWED_DATASETS:\n",
+    "    raise ValueError(f\"Dataset {dataset!r} is not allowed. Allowed: {', '.join(sorted(ALLOWED_DATASETS))}\")\n",
+    "\n",
+    "state_path = Path(os.environ[\"SYFT_POLICY_DEFAULT_STATE_PATH\"])\n",
+    "submitter = os.environ[\"SYFT_POLICY_SUBMITTER\"]\n",
+    "state = json.loads(state_path.read_text())\n",
+    "\n",
+    "now = datetime.now(timezone.utc)\n",
+    "calls = [t for t in state.get(submitter, []) if now - datetime.fromisoformat(t) < timedelta(days=1)]\n",
+    "if len(calls) >= MAX_CALLS:\n",
+    "    raise ValueError(f\"Rate limit reached: {MAX_CALLS} calls per 24 hours\")\n",
+    "\n",
+    "state[submitter] = calls + [now.isoformat()]\n",
+    "state_path.write_text(json.dumps(state))"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "id": "47",
+   "metadata": {
+    "id": "api-policy-test-md"
+   },
+   "source": [
+    "Test it before you attach it. `syft_bg.run_policy` runs a policy the way the runner does, with no job, peer or sync. `error` is what the DS would see."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "48",
+   "metadata": {
+    "id": "api-policy-test-cell"
+   },
+   "outputs": [],
+   "source": [
+    "import tempfile\n",
+    "\n",
+    "ok = syft_bg.run_policy(\"input_policy.py\", params={\"dataset\": \"ocean_water_quality2\"})\n",
+    "refused = syft_bg.run_policy(\"input_policy.py\", params={\"dataset\": \"lab_reports3\"})\n",
+    "print(f\"ocean_water_quality2: passed={ok.passed}\")\n",
+    "print(f\"lab_reports3: passed={refused.passed}, the DS sees {refused.error!r}\")\n",
+    "\n",
+    "# Four calls against one state folder: the fourth is over the limit.\n",
+    "state_dir = tempfile.mkdtemp()\n",
+    "for n in range(1, 5):\n",
+    "    r = syft_bg.run_policy(\"input_policy.py\", params={\"dataset\": \"ocean_water_quality2\"}, state_dir=state_dir)\n",
+    "    print(f\"call {n}: passed={r.passed}\", r.error or \"\")"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "id": "49",
+   "metadata": {
+    "id": "api-policy-attach-md"
+   },
+   "source": [
+    "Attach it to the api you created above. From the next call on, a refused call is `failed`, and the DS reads your message in `job.policy_error`. `set_policies` finds your apis through the `syft_bg.init` config from Step 10.\n",
+    "\n",
+    "> 💡 When you create an api, you can attach a policy straight away: `syft_bg.create_api_from_job(job, input_policy=\"input_policy.py\")`."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "50",
+   "metadata": {
+    "id": "api-policy-attach-cell"
+   },
+   "outputs": [],
+   "source": [
+    "attached = syft_bg.set_policies(result.name, input_policy=\"input_policy.py\")\n",
+    "if attached.success:\n",
+    "    print(f\"✅  Input policy attached to {result.name!r}\")\n",
+    "else:\n",
+    "    print(f\"❌  Failed: {attached.error}\")"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "id": "51",
    "metadata": {
     "id": "step12-md"
    },
@@ -1017,7 +1138,7 @@
   {
    "cell_type": "code",
    "execution_count": null,
-   "id": "46",
+   "id": "52",
    "metadata": {
     "id": "bg-logs-cell"
    },
@@ -1036,7 +1157,7 @@
   },
   {
    "cell_type": "markdown",
-   "id": "47",
+   "id": "53",
    "metadata": {
     "id": "step13-md"
    },
@@ -1054,7 +1175,7 @@
   {
    "cell_type": "code",
    "execution_count": null,
-   "id": "48",
+   "id": "54",
    "metadata": {
     "id": "reset-bg-cell"
    },
@@ -1063,13 +1184,13 @@
     "# Uncomment ONLY if you want to wipe syft-bg state and start fresh:\n",
     "\n",
     "# syft_bg.reset()\n",
-    "# print(\"✅ All services stopped and uninstalled. Re-run Step 11 to start again.\")\n"
+    "# print(\"✅ All services stopped and uninstalled. Re-run Step 10 to start again.\")\n"
    ]
   },
   {
    "cell_type": "code",
    "execution_count": null,
-   "id": "49",
+   "id": "55",
    "metadata": {
     "id": "delete-dataset-cell"
    },
@@ -1082,7 +1203,7 @@
   },
   {
    "cell_type": "markdown",
-   "id": "50",
+   "id": "56",
    "metadata": {
     "id": "done-md"
    },
@@ -1098,6 +1219,7 @@
     "- ✅ Ran the job and shared the results back\n",
     "- ✅ Started background services for unattended operation\n",
     "- ✅ Created an api from a reviewed job, so the DS can call it without a new review\n",
+    "- ✅ Added an input policy that checks each call before it runs\n",
     "\n",
     "## Day-to-day workflow\n",
     "\n",
@@ -1110,7 +1232,7 @@
     "# Or, in the notebook: client.process_approved_jobs(share_outputs_with_submitter=True)\n",
     "```\n",
     "\n",
-    "If something gets stuck, `syft_bg.reset()` + re-run Step 11 brings up a clean state.\n",
+    "If something gets stuck, `syft_bg.reset()` + re-run Step 10 brings up a clean state.\n",
     "\n",
     "## Resources\n",
     "\n",

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

@@ -36,10 +36,16 @@
                 file in SYFT_POLICY_STATE_DIR. Nothing changes unless the policy
                 writes. The runner holds a lock on the state for the whole run,
                 so two calls of one api never run their policies at once.
-    Deny        Exit with a non-zero code. In Python, any uncaught exception
-                does this. The job is marked failed.
+    Describe    Start the policy with a one- or two-line docstring (a comment
+                in bash) that says in plain words what it allows and refuses.
+                The data scientist sees the policy's full source in client.api.
+    Deny        Raise an exception, or exit with a non-zero code. The job is
+                marked failed.
     Reason      The last line of stderr is what the data scientist sees, in
-                policy_error.txt. The full stderr stays with the data owner.
+                policy_error.txt. For an uncaught exception that is its name
+                and message, e.g. "ValueError: Rate limit reached", so keep
+                the message to one line. The full stderr stays with the data
+                owner.
     Call files  A call sends exactly the files the api lists by name. The
                 input policy can read them in SYFT_POLICY_CALL_DIR and deny the
                 call because of what they hold.