diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 6f6e293..423689f 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -6,6 +6,14 @@ on: pull_request: jobs: + conflict-markers: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + + - name: Check for merge conflict markers + run: "! git grep -nE '^(<<<<<<<|>>>>>>>)( |$)|^=======$'" + test: runs-on: ubuntu-latest strategy: @@ -26,4 +34,28 @@ jobs: run: pip install -e ".[dev]" - name: Run fast test suite (no network, no Docker, no FLARE simulator) - run: python -m pytest -q -m "not slow" \ No newline at end of file + run: python -m pytest -q -m "not slow" + + frontend: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + + - name: Set up Node.js + uses: actions/setup-node@v4 + with: + node-version: "24" + cache: npm + cache-dependency-path: frontend/package-lock.json + + - name: Install dependencies + working-directory: frontend + run: npm ci + + - name: Run tests + working-directory: frontend + run: npm test + + - name: Build + working-directory: frontend + run: npm run build diff --git a/AGENTS.md b/AGENTS.md index d42ef5d..b7d7a1e 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -26,10 +26,8 @@ Implemented: | Server side | `server/aggregate.py` (associative merge), `server/disclosure_check.py`, `server/overseer_queue.py` (CLI); outputs under `server/out/` (gitignored, `SERVER_OUT` env) | | Wire contracts | `adapters/base.py` (`AggregateResult`, `VariableStats`, `Gram` — the TRE→server payload, strict Pydantic), `server/schemas.py` (`MergedResult`, `ReleasedResult`, HTTP API models); JSON Schema + OpenAPI in `docs/schemas/` via `scripts/export_schemas.py` (a test fails when stale) | | HTTP API + UI | `server/api.py` serves the researcher UI contract (`/metadata`, `/examples/{name}`, `POST /run`, `/run/{id}`, `/overseer`, `/overseer/{hash}/{approve,reject}`, `/audit`) and the convenience endpoints (`/allele-frequency`, `/linear-regression`, `/logistic-regression`); `frontend/` (Next.js) is wired to it — run API on :8500, UI on :3000 | -| Ops scripts | `scripts/local_federation.sh` (real FLARE, no Docker), `scripts/provision.sh`, `scripts/onboard_tre.sh`, `scripts/start_server.sh`, `scripts/start_client.sh`, `scripts/run_job.py` (`--mode simulator|prod`, `--fedavg`), `scripts/run_local.py`, `scripts/verify.py`, `scripts/scale_sim.py` → `docs/scaling.png` | -| Tests | `tests/` — 111 fast + 4 FLARE-simulator (`-m slow`) | - -A real (non-simulator) FLARE federation runs two verified ways: `scripts/local_federation.sh` with separate processes over mTLS, and Docker via `scripts/provision.sh && scripts/up.sh --flare`; both support the example specs, overseer flow, and partial-site coverage. The Next.js UI is implemented in `frontend/`; the Python API service exposing its documented contract remains separate. Not yet run: clients on remote hosts (Gefion / NextCloud). +| Ops scripts | `scripts/local_federation.sh` (real FLARE, no Docker), `scripts/provision.sh`, `scripts/limit_server_jobs.py` (both provisioning scripts run it: server kit runs one job at a time, `max_jobs: 1`), `scripts/onboard_tre.sh`, `scripts/start_server.sh`, `scripts/start_client.sh`, `scripts/run_job.py` (`--mode simulator|prod`, `--fedavg`), `scripts/run_local.py`, `scripts/verify.py`, `scripts/scale_sim.py` → `docs/scaling.png` | +| Tests | `tests/` — 149 fast + 4 FLARE-simulator (`-m slow`); `frontend/test/` (`npm test`) | `docs/architecture/flowchart.drawio` is the editable source for the implemented-flow overview. When changing that overview, regenerate both `docs/architecture/flowchart_drawio.svg` and `docs/architecture/flowchart.png` from it. @@ -45,7 +43,7 @@ Confirm a path, command, module, or function exists before referencing it — th Checks that exist and were confirmed working on 2026-09-17: -- `python -m pytest -q -m "not slow"` — 111 passed, ~4 s, no network. +- `python -m pytest -q -m "not slow"` — 149 tests as of 2026-09-25, ~6 s, no network. - `python -m pytest -q -m slow` — 4 passed, ~75 s; starts the TREs as local processes and runs FLARE simulator jobs (allele_freq, straggler, FedAvg linreg, Newton-Raphson logreg). - `scripts/dev_tres.py` + `scripts/run_job.sh spec/examples/.json` — simulator end-to-end with `scripts/verify.py` against ground truth. - `scripts/local_federation.sh up && scripts/local_federation.sh job spec/examples/allele_freq.json` — real FLARE server + 3 clients on localhost; ~12 s per job. @@ -53,7 +51,7 @@ Checks that exist and were confirmed working on 2026-09-17: - `scripts/onboard_tre.sh ` — adds a site, regenerates, provisions, packs a kit (~2 s). - `scripts/provision.sh && scripts/up.sh --flare` — builds 7 images (~5 min cold), starts everything, checks each TRE's `/health` from its own FLARE client container, asserts no TRE container can reach the public internet. Then `docker compose exec flare-server python scripts/run_job.py --mode prod --admin-kit "/workspace/federated_apis/prod_00/admin@ncfh.org" spec/examples/allele_freq.json` (~11 s) and `python scripts/verify.py server/out//result.json` on the host. Verified 2026-09-17 with Docker Desktop 29.8 on macOS. -CI: `.github/workflows/ci.yml` runs `python -m pytest -q -m "not slow"` on every push and pull request, against Python 3.11 and 3.12. No linter or formatter is configured. +CI: `.github/workflows/ci.yml` runs on every push and pull request: a merge-conflict-marker check over tracked files, `python -m pytest -q -m "not slow"` against Python 3.11 and 3.12, and a frontend job (Node 24: `npm ci`, `npm test`, `npm run build`). No linter or formatter is configured. For documentation changes, check source accuracy, links, and `git diff --check`. diff --git a/README.md b/README.md index e2750aa..6040145 100644 --- a/README.md +++ b/README.md @@ -11,7 +11,7 @@ CI Python 3.11+ NVIDIA FLARE 2.9 - tests + tests MIT

@@ -57,7 +57,7 @@ cd frontend && npm install && npm run dev # Next.js UI The fast suite needs no network; the FLARE simulator suite is marked `slow`: ```sh -python -m pytest -q -m "not slow" # 111 tests, ~3 s +python -m pytest -q -m "not slow" # 149 tests, ~6 s python -m pytest -q -m slow # 4 tests, ~70 s ``` @@ -70,7 +70,7 @@ python scripts/dev_tres.py python scripts/run_local.py spec/examples/allele_freq.json ``` -**FLARE simulator** — one process, real controllers and executors. Each run writes `server/out//result.json`; `released.json` appears only when the disclosure check passes or an overseer approves, and is what would leave the server. Directories are keyed by spec hash, so an earlier `released.json` can outlive a later flagged run of the same spec — `check.json` holds the current decision. +**FLARE simulator** — one process, real controllers and executors. Each run writes `server/out//result.json`; `released.json` appears only when the disclosure check passes or an overseer approves, and is what would leave the server. Directories are keyed by spec hash and describe the latest run: a flagged rerun withdraws an earlier `released.json`, a released rerun clears a stale queue entry, and the release log keeps the history. ```sh scripts/run_job.sh spec/examples/allele_freq.json @@ -102,9 +102,9 @@ scripts/up.sh --down 1. **Request** — a JSON `AnalysisSpec` in canonical variable names; hashed, and that hash keys every log. 2. **Dispatch** — the FLARE server sits outside every TRE; clients dial **out** over mTLS gRPC. No inbound ports in a secure environment. 3. **Local execution** — the FLARE client loads the adapter for its own `tre_id` and speaks the TRE's native API (REST, DataSHIELD-style, SQL gateway). Adapters can ask only for schema metadata and five aggregate primitives: `count`, `describe`, `value_counts`, `gram`, `irls_step`. -4. **Safe output** — inside the TRE, before the FLARE client sees anything: project allow-list, aggregate allow-list, `n < k` ⇒ nothing leaves, cell `< k` suppressed (and anything derivable from it), one audit line per request and per round. +4. **Safe output** — in the adapter, before anything leaves it: inside the TRE on the FLARE path (on the direct-API demo path the adapter runs in the API server process, so unfiltered aggregates reach that process first). Project allow-list, aggregate allow-list, `k` = the request's `min_cell_size` raised to the project's floor, `n < k` ⇒ nothing leaves, a variable observed in fewer than `k` rows ⇒ nothing about it leaves, cell `< k` suppressed (and anything derivable from it), regression inputs must be complete (a missing value rejects the site), one audit line per request and per round, accepted or rejected. 5. **Aggregation** — everything the server merges is a sum (counts, Gram matrices, gradients and Hessians, n-weighted β), so merging is exact, order-free and straggler-tolerant: `min_clients` + `wait_time`, never wait-for-all; a missing site is reported as `2/3 sites`. -6. **Disclosure check and release** — on the merged table: minimum cell size, dominance, ≥ 2 sites, site-level suppressions, differencing against earlier releases. Pass ⇒ `released.json`; flag ⇒ overseer queue, human decision logged. +6. **Disclosure check and release** — on the merged table: minimum cell size, dominance (per table cell, per variable's observed values, per regression sample and per released federated round), ≥ 2 sites (overall and in every released round), site-level suppressions, differencing against earlier releases. Pass ⇒ `released.json`; flag ⇒ overseer queue, human decision logged. [Edit the diagram](docs/architecture/flowchart.drawio) · [Original external diagram](https://drive.google.com/file/d/1j9t8W-cFVBYgHGrFrLGP5keU2vaFtE-l/view?usp=sharing) diff --git a/adapters/base.py b/adapters/base.py index f0e6c65..7762eb3 100644 --- a/adapters/base.py +++ b/adapters/base.py @@ -115,13 +115,15 @@ def value_counts(self, col: str, filters: dict) -> dict[str, int]: @abstractmethod def gram(self, cols: list[str], filters: dict) -> dict: - """Cross-product matrix of [1, *cols]: {"n", "cols", "matrix"}; matrix[i][j] = sum(x_i * x_j).""" + """Cross-product matrix of [1, *cols]: {"n", "cols", "matrix", "complete"}; matrix[i][j] = sum(x_i * x_j). + `complete` is False when any row misses any of `cols`; the matrix is then meaningless (None).""" @abstractmethod def irls_step(self, outcome_col: str, feature_cols: list[str], beta: list[float], filters: dict) -> dict: """One Newton-Raphson step's worth of sufficient statistics for logistic regression on [1, *feature_cols] -> outcome_col, evaluated at beta: - {"n", "grad" (len p+1), "hess" ((p+1)x(p+1))}.""" + {"n", "complete", "grad" (len p+1), "hess" ((p+1)x(p+1))}; grad/hess are + None when `complete` is False (some row misses the outcome or a feature).""" # ---- shared logic ---------------------------------------------------------------- def local(self, canonical: str) -> str: @@ -137,8 +139,8 @@ def run(self, spec: AnalysisSpec) -> AggregateResult: filters = self.local_filters(spec) n = self.count(filters) raw: dict[str, dict[str, Any]] = {} - if n < spec.min_cell_size: # don't even compute on a cohort that can't be released - return safe_output.filter(self.tre_id, self.region, spec, n, raw) + if n < safe_output.effective_min_cell_size(spec.project_id, spec.min_cell_size): + return safe_output.filter(self.tre_id, self.region, spec, n, raw) # don't compute on an unreleasable cohort if spec.analysis_type in ("allele_freq", "fed_stats"): for v in spec.variables: @@ -151,6 +153,8 @@ def run(self, spec: AnalysisSpec) -> AggregateResult: elif spec.analysis_type == "fed_linreg": cols = [spec.outcome, *spec.variables] g = self.gram([self.local(c) for c in cols], filters) + if g.get("complete") is not True: # missing values: reject rather than fit a silently different cohort + return safe_output.reject_all(self.tre_id, self.region, spec, reason="_linreg.gram:incomplete_inputs") raw["_linreg"] = {"gram": {"n": g["n"], "cols": ["intercept", *cols], "matrix": g["matrix"]}} return safe_output.filter(self.tre_id, self.region, spec, n, raw) diff --git a/adapters/rest.py b/adapters/rest.py index 5aee9f6..147b4b5 100644 --- a/adapters/rest.py +++ b/adapters/rest.py @@ -29,4 +29,4 @@ def gram(self, cols: list[str], filters: dict) -> dict: def irls_step(self, outcome_col: str, feature_cols: list[str], beta: list[float], filters: dict) -> dict: r = self._post("/irls", {"outcome": outcome_col, "features": feature_cols, "beta": beta, "filters": filters}) - return {"n": r["n"], "grad": r["grad"], "hess": r["hess"]} + return {"n": r["n"], "complete": r.get("complete"), "grad": r.get("grad"), "hess": r.get("hess")} diff --git a/adapters/safe_output.py b/adapters/safe_output.py index e3e99fc..ba3bd81 100644 --- a/adapters/safe_output.py +++ b/adapters/safe_output.py @@ -1,17 +1,27 @@ -"""Safe Output filter -- runs INSIDE the TRE, before anything reaches the FLARE client. +"""Safe Output filter -- runs wherever the adapter runs: inside the TRE on the FLARE +path (the executor calls the adapter), but inside the API server process on the +direct-API demo path (server/analysis_service.py), so there the unfiltered +aggregates reach the server before this filter applies. Rules (all decisions are appended to audit.jsonl): 1. project_id must be on the Safe Projects allow-list (projects.yaml) 2. only allow-listed aggregate keys leave; anything else is dropped - 3. if the filtered cohort has n < min_cell_size, nothing leaves - 4. any count cell < min_cell_size is suppressed; if a genotype cell is + 3. k = max(requested min_cell_size, the project's floor in projects.yaml); + if the filtered cohort has n < k, nothing leaves + 4. any count cell < k is suppressed; if a genotype cell is suppressed, derived allele counts for that variable are withheld too - (they would let the cell be back-calculated) - 5. a regression Gram matrix leaves only if n >= max(min_cell_size, #cols + 1) + (they would let the cell be back-calculated); a variable observed in + fewer than k rows (scalar count < k, or a table with no observed level) + releases nothing at all + 5. a regression Gram matrix leaves only if n >= max(k, #cols + 1), and only + if every row has every regression column (no missing values) + 6. a Newton-Raphson round (gradient + Hessian) leaves under the same two + conditions, via release_irls_step """ from __future__ import annotations import json +import math import os import time from pathlib import Path @@ -27,19 +37,33 @@ ALLOWED_SCALARS = {"count", "sum", "sum_sq", "min", "max"} ALLOWED_TABLES = {"genotype_counts", "histogram"} ALLOWED_MATRIX = {"gram"} +DEFAULT_MIN_CELL_SIZE = 5 # floor for a project that does not set its own -def project_allowed(project_id: str, tre_id: str) -> bool: +class Rejected(Exception): + """A release the Safe Output rules refused; already audited.""" + + +def _projects() -> dict: if not PROJECTS_PATH.exists(): - return False - projects = yaml.safe_load(PROJECTS_PATH.read_text()).get("projects") or {} - p = projects.get(project_id) + return {} + return yaml.safe_load(PROJECTS_PATH.read_text()).get("projects") or {} + + +def project_allowed(project_id: str, tre_id: str) -> bool: + p = _projects().get(project_id) if p is None: return False sites = p.get("sites", "all") return sites == "all" or tre_id in sites +def effective_min_cell_size(project_id: str | None, requested: int) -> int: + """The researcher may ask for a larger k, never a smaller one than policy allows.""" + floor = (_projects().get(project_id) or {}).get("min_cell_size", DEFAULT_MIN_CELL_SIZE) + return max(int(requested), int(floor)) + + def audit_path(tre_id: str) -> Path: d = Path(os.environ.get("AUDIT_DIR", ROOT / "audit" / tre_id)) d.mkdir(parents=True, exist_ok=True) @@ -55,7 +79,7 @@ def _audit(tre_id: str, spec: AnalysisSpec, n: int, released: dict, rejected: li "analysis_type": spec.analysis_type, "variables": spec.variables + ([spec.outcome] if spec.outcome else []), "filters": spec.filters, - "min_cell_size": spec.min_cell_size, + "min_cell_size": effective_min_cell_size(spec.project_id, spec.min_cell_size), "n": n, "released": {v: sorted(s.keys()) for v, s in released.items()}, "rejected": rejected, @@ -65,6 +89,17 @@ def _audit(tre_id: str, spec: AnalysisSpec, n: int, released: dict, rejected: li f.write(json.dumps(entry) + "\n") +def _observed(stats: dict[str, Any]) -> int | None: + """Rows in which a variable was observed, where that decides the whole variable: + a scalar variable's count, or 0 for a table without a single observed level + (a table with levels is checked cell by cell).""" + if stats.get("count") is not None: + return int(stats["count"]) + if any(key in ALLOWED_TABLES and not table for key, table in stats.items()): + return 0 + return None + + def reject_all(tre_id: str, region: str, spec: AnalysisSpec, reason: str): from adapters.base import AggregateResult # local import: base imports this module @@ -75,7 +110,7 @@ def reject_all(tre_id: str, region: str, spec: AnalysisSpec, reason: str): def filter(tre_id: str, region: str, spec: AnalysisSpec, n: int, raw: dict[str, dict[str, Any]]): # noqa: A001 from adapters.base import AggregateResult - k = spec.min_cell_size + k = effective_min_cell_size(spec.project_id, spec.min_cell_size) rejected: list[str] = [] out: dict[str, dict[str, Any]] = {} @@ -85,6 +120,11 @@ def filter(tre_id: str, region: str, spec: AnalysisSpec, n: int, raw: dict[str, return AggregateResult(tre_id=tre_id, n=0, stats={}, rejected=rejected, region=region, spec_hash=spec.spec_hash()) for var, stats in raw.items(): + observed = _observed(stats) + if observed is not None and observed < k: # e.g. one person's value in an otherwise large cohort + rejected.append(f"{var}.*:observed={observed}<{k}") + out[var] = {} + continue kept: dict[str, Any] = {} for key, val in stats.items(): if key in ALLOWED_SCALARS: @@ -126,3 +166,31 @@ def audit_release(tre_id: str, spec: AnalysisSpec, n: int, released: dict[str, l """Audit something derived inside the TRE from an already-filtered result (e.g. per-round model parameters in federated learning).""" _audit(tre_id, spec, n, {k: {x: None for x in v} for k, v in released.items()}, [], f"OK:{note}") + + +def release_irls_step(tre_id: str, spec: AnalysisSpec, step: Any, p: int) -> dict: + """Safe Output for one Newton-Raphson round from one site: the adapter's + irls_step answer (n, complete, grad, hess) at a p-parameter beta. Returns + only {n, grad, hess}; raises Rejected (after auditing) otherwise. Same + sample rule as a Gram matrix over [intercept, outcome, *features].""" + n = step.get("n") if isinstance(step, dict) else None + + def reject(reason: str): + _audit(tre_id, spec, n if type(n) is int else 0, {}, [f"_logreg.step:{reason}"], "REJECTED") + raise Rejected(reason) + + if type(n) is not int or n < 0: + reject("invalid_n") + if step.get("complete") is not True: + reject("incomplete_inputs") + grad, hess = step.get("grad"), step.get("hess") + if (not isinstance(grad, list) or len(grad) != p or not isinstance(hess, list) or len(hess) != p + or any(not isinstance(row, list) or len(row) != p for row in hess)): + reject("invalid_shape") + if any(type(v) not in (int, float) or not math.isfinite(v) for v in [*grad, *(v for row in hess for v in row)]): + reject("non_finite") + needed = max(effective_min_cell_size(spec.project_id, spec.min_cell_size), p + 2) + if n < needed: + reject(f"n={n}<{needed}") + audit_release(tre_id, spec, n, {"_logreg": ["grad", "hess"]}, "newton_raphson_round") + return {"n": n, "grad": grad, "hess": hess} diff --git a/adapters/sql.py b/adapters/sql.py index 39f28e4..e9a4d84 100644 --- a/adapters/sql.py +++ b/adapters/sql.py @@ -66,14 +66,19 @@ def gram(self, cols: list[str], filters: dict) -> dict: raise ValueError(f"bad column {c!r}") terms = ["1", *cols] exprs = [f"sum(({a})*({b}))" for a in terms for b in terms] - row = self._sql(f"SELECT count(*), {', '.join(exprs)} FROM cohort{where(filters)}")[0] - k = len(terms) - vals = [float(v) for v in row[1:]] - return {"n": int(row[0]), "cols": terms, "matrix": [vals[i * k:(i + 1) * k] for i in range(k)]} + # SUM skips NULLs term by term, so completeness is counted in the same query: + # an incomplete cohort's sums would each cover a different set of rows. + complete = f"count(CASE WHEN {' AND '.join(f'{c} IS NOT NULL' for c in cols)} THEN 1 END)" + row = self._sql(f"SELECT count(*), {complete}, {', '.join(exprs)} FROM cohort{where(filters)}")[0] + n, k = int(row[0]), len(terms) + if int(row[1]) != n: + return {"n": n, "cols": terms, "matrix": None, "complete": False} + vals = [float(v) for v in row[2:]] + return {"n": n, "cols": terms, "matrix": [vals[i * k:(i + 1) * k] for i in range(k)], "complete": True} def irls_step(self, outcome_col: str, feature_cols: list[str], beta: list[float], filters: dict) -> dict: # Nonlinear per-row transform parameterised by beta -- not expressible as one # aggregating SELECT, so this goes through the gateway's dedicated /irls endpoint # instead of the generic SQL passthrough (same trust boundary as /schema). r = self._post("/irls", {"outcome": outcome_col, "features": feature_cols, "beta": beta, "filters": filters}) - return {"n": r["n"], "grad": r["grad"], "hess": r["hess"]} + return {"n": r["n"], "complete": r.get("complete"), "grad": r.get("grad"), "hess": r.get("hess")} diff --git a/docs/MANUAL_API_TEST.md b/docs/MANUAL_API_TEST.md index 709da35..b181204 100644 --- a/docs/MANUAL_API_TEST.md +++ b/docs/MANUAL_API_TEST.md @@ -90,7 +90,7 @@ Expect Uvicorn listening on `http://127.0.0.1:8500`. The API can start and serve `SERVER_OUT` isolates disclosure decisions, overseer records and release history. `AUDIT_DIR` isolates adapter audit records; `AUDIT_ROOT` keeps `/audit` from also reading older repository audit files. These temporary directories contain demo records, **not the Python environment**. A fresh directory prevents previous runs' disclosure history from affecting a new demonstration. -Keep `--workers 1`: release coordination uses a process-local lock and background run status is held in memory. No other environment variables need to be set for this procedure. Optional `API_TRE_TIMEOUT` and `API_TRE_WORKERS` default to 10 seconds per HTTP operation and 8 workers. +Keep `--workers 1`: release coordination and overseer decisions use a process-local lock and background run status is held in memory. No other environment variables need to be set for this procedure. Optional `API_TRE_TIMEOUT` and `API_TRE_WORKERS` default to 10 seconds per HTTP operation and 8 workers. ## 5. Swagger @@ -135,14 +135,14 @@ These routes also exist in [server/api.py](../server/api.py). Keep them separate | --- | --- | | `POST /logistic-regression` | Conditional: use features `age`, `bmi`, target `case`, the same project and empty filters. Defaults: `rounds=25`, `tol=1e-8`, `min_cell_size=5`. Requires local outcome columns `case_status` (HUNT), `is_case` (Gefion), `CASE_STATUS` (Brev). The inspected CSVs lack these; regenerate deliberately with the current generator and restart TREs first. Not part of the basic demo. | | `GET /examples/{name}` | Suitable, read-only; e.g. `/examples/allele_freq.json` and `/examples/fed_linreg.json`. Returns a full `AnalysisSpec`, not the convenience endpoint's request schema. No data generation needed. | -| `POST /run` | Optional background submission of a full `AnalysisSpec`; returns HTTP 202 and a `run_id`. Use an allele-frequency or linear-regression example for existing data. Optional `fedavg_rounds` applies only to linear regression; omit it for the exact solve. Avoid `fed_stats` in this demo: the current service's final statistics validation assumes a regression fit for non-allele analyses. | +| `POST /run` | Optional background submission of a full `AnalysisSpec`; returns HTTP 202 and a `run_id`. Use an allele-frequency or linear-regression example for existing data. Optional `fedavg_rounds` applies only to linear regression; omit it for the exact solve. `fed_stats` also works, e.g. with variables `age` and `bmi`. | | `GET /run/{run_id}` | Suitable for polling a submitted run. Run status is in memory and disappears on API restart; no extra data needed. | | `GET /overseer` | Optional internal governance view. May be empty after successful requests. Contains detailed internal disclosure reasons; use only with local synthetic demo data. | -| `POST /overseer/{spec_hash}/{decision}` | Administrative mutation, with `decision` equal to `approve` or `reject` and optional `note`/`by` body. Skip in the core demo. It changes stored decisions; already cached run status does not refresh automatically. | +| `POST /overseer/{spec_hash}/{decision}` | Administrative mutation, with `decision` equal to `approve` or `reject` and optional `note`/`by` body. Skip in the core demo. It changes stored decisions; runs of that spec still flagged in memory then report `completed` with decision `APPROVED` and the released result, or `rejected` with decision `REJECTED`. | | `GET /audit` | Optional read-only internal audit view after requests. No extra data needed; contains internal records, not the sanitized researcher response. | | `GET /redoc`, `GET /openapi.json` | Suitable documentation/schema views; no TREs or data required. | -The governance endpoints currently have no authentication layer. Keep this procedure on localhost with synthetic data. +The governance endpoints currently have no authentication layer. `GET /overseer` and `GET /audit` show pre-suppression detail meant for overseers and auditors only — exact small counts behind each suppression, per-site cohort sizes of rejected requests, dominant sites — so anyone who can reach the API sees it. Keep this procedure on localhost with synthetic data; authentication is a disclosure requirement before any shared deployment. ## What this validates: direct adapter communication diff --git a/docs/demo_specs.md b/docs/demo_specs.md index c995a73..c083cce 100644 --- a/docs/demo_specs.md +++ b/docs/demo_specs.md @@ -2,7 +2,7 @@ Four example analysis specifications live in `spec/examples/`. Together they walk through the full path — one request, three heterogeneous TREs, local computation, safe output, server-side aggregation and disclosure control — and they cover the release case, the filtered case, the suppression case and the federated-learning case. -Every spec is an `AnalysisSpec` (`spec/analysis_spec.py`): canonical variable names only, a `project_id` that must appear in the Safe Projects allow-list (`projects.yaml`), and a `min_cell_size` that sets the suppression threshold. The spec hash keys the audit and release logs. +Every spec is an `AnalysisSpec` (`spec/analysis_spec.py`): canonical variable names only, a `project_id` that must appear in the Safe Projects allow-list (`projects.yaml`), and a `min_cell_size` that sets the suppression threshold — raised to the project's `min_cell_size` floor in `projects.yaml` (default 5), never lowered below it. The spec hash keys the audit and release logs. ## Running them @@ -49,7 +49,7 @@ This spec is not exactness-checked. When the `spec.json` beside the result conta **What should happen.** This is a *partial* release, not a total rejection. Verified by `tests/test_m2_adapters.py::test_small_cells_suppressed_and_audited`: at every site the homozygous-minor genotype cell (`2`) falls below 100 and is withheld, and because allele counts could be back-calculated from the remaining cells, they are withheld too — `snp_rs001.allele_counts:derived_from_suppressed_cell`. Each site audits the run as `PARTIAL`. -Because every site suppressed something, the server-side check adds a `site_suppression` reason per site (`server/disclosure_check.py`), so the merged result is `FLAGGED`. This flagged result requires overseer approval before release, via `server/overseer_queue.py`. An older `released.json` can remain from a previous run of the same spec, since run directories are keyed by spec hash — `check.json` carries the current decision. +Because every site suppressed something, the server-side check adds a `site_suppression` reason per site (`server/disclosure_check.py`), so the merged result is `FLAGGED`. This flagged result requires overseer approval before release, via `server/overseer_queue.py`. Run directories are keyed by spec hash and describe the latest run: a flagged rerun withdraws an earlier `released.json`, and the release log keeps the history. **What it demonstrates.** How the disclosure controls handle small cells end to end: a local rule fires, the reason is recorded in each TRE's audit log, the server notices the incomplete table, and release waits on a person. @@ -57,7 +57,7 @@ Because every site suppressed something, the server-side check adds a `site_supp **Purpose.** Federated linear regression: `sbp` on age, sex, BMI, LDL and the three causal variants. -**What should happen.** Each TRE returns a Gram matrix — the sufficient statistic for OLS — and nothing else; the server sums them and solves once. The filter releases a Gram matrix only when `n >= max(min_cell_size, #cols + 1)`. That is a sample-count threshold, not a test for near-singularity: it withholds output from a site with too few rows to fit the model, but it does not detect an ill-conditioned one. +**What should happen.** Each TRE returns a Gram matrix — the sufficient statistic for OLS — and nothing else; the server sums them and solves once. The filter releases a Gram matrix only when `n >= max(min_cell_size, #cols + 1)` and every selected row has every regression column; a site with missing values is rejected rather than fitted on a different set of rows. That is a sample-count threshold, not a test for near-singularity: it withholds output from a site with too few rows to fit the model, but it does not detect an ill-conditioned one. A FedAvg variant runs through `scripts/run_job.py --fedavg`. Rows still never leave, but more than coefficients does: in round 0 each site reports its `n` plus feature sums and sums of squares, so the server can fix one global standardisation; in each later round a site returns its `n` and its updated coefficients (`flare/app/linreg.py`). @@ -71,4 +71,3 @@ Read this before quoting any of it as a result. - **Verified on one machine.** That run was a real provisioned FLARE federation over mTLS, not the simulator, but every process was local. Running in Docker containers (`scripts/provision.sh && scripts/up.sh --flare`) was verified separately on 2026-09-17 (see [AGENTS.md](../AGENTS.md)), but the expected outcomes here were not re-captured under Docker. Running across machines or institutions is still unverified. - **`verify.py` checks what is present.** It reads the merged result, not the released artifact. It compares allele frequencies even under partial coverage, and means and OLS coefficients only when every site reported. Statistics the filter suppressed are skipped entirely. `PASS` means no compared statistic exceeded tolerance — zero comparisons still produce `PASS`. - **Synthetic data, invented column names.** The cohort comes from `data/generate.py` and the per-site local names in `docs/variables.md` are placeholders for the mock TREs, agreed with no real site. -- **`released.json` can be stale.** Run directories are keyed by spec hash, so a release from an earlier run of the same spec survives a later flagged run. `check.json` carries the current decision. diff --git a/docs/schemas/openapi.json b/docs/schemas/openapi.json index c1f7e3a..52cef84 100644 --- a/docs/schemas/openapi.json +++ b/docs/schemas/openapi.json @@ -1602,6 +1602,7 @@ "running", "completed", "flagged", + "rejected", "failed" ], "title": "Status" @@ -1638,7 +1639,9 @@ "type": "string", "enum": [ "OK", - "FLAGGED" + "FLAGGED", + "APPROVED", + "REJECTED" ] }, { diff --git a/docs/schemas/run_status.json b/docs/schemas/run_status.json index 13953d9..14b2000 100644 --- a/docs/schemas/run_status.json +++ b/docs/schemas/run_status.json @@ -602,6 +602,7 @@ "running", "completed", "flagged", + "rejected", "failed" ], "title": "Status", @@ -640,7 +641,9 @@ { "enum": [ "OK", - "FLAGGED" + "FLAGGED", + "APPROVED", + "REJECTED" ], "type": "string" }, diff --git a/docs/variables.md b/docs/variables.md index f7cd08c..d389803 100644 --- a/docs/variables.md +++ b/docs/variables.md @@ -29,7 +29,7 @@ Twenty variants, `snp_rs001` through `snp_rs020`, each a minor-allele dosage of | `gefion` | `SNP_` | `SNP_1` | | `brev` | `g_rs` | `g_rs001` | -The `rs001`–`rs020` identifiers are synthetic placeholders, not real dbSNP rsIDs. Allele frequency is computed as `sum(dosage) / (2 × n)`; that denominator fixes ploidy at 2 and says nothing about orientation. Which allele the dosage counts is set by the schema and the generator: `harmonisation/canonical.yaml` describes every column as minor-allele dosage, and the generator draws `Binomial(2, p)` with `p ≤ 0.5` at every site, so the minor allele stays the minor allele throughout the synthetic cohort. Real cohorts give no such guarantee. The synthetic data contains no missing values, so a missing-value encoding has never been exercised. +The `rs001`–`rs020` identifiers are synthetic placeholders, not real dbSNP rsIDs. Allele frequency is computed as `sum(dosage) / (2 × n)`; that denominator fixes ploidy at 2 and says nothing about orientation. Which allele the dosage counts is set by the schema and the generator: `harmonisation/canonical.yaml` describes every column as minor-allele dosage, and the generator draws `Binomial(2, p)` with `p ≤ 0.5` at every site, so the minor allele stays the minor allele throughout the synthetic cohort. Real cohorts give no such guarantee. The synthetic data contains no missing values. Regression analyses reject a site whose selected rows miss any regression column (tested with blanked values); a missing-value encoding in real source data has never been exercised. ## Synthetic cohort diff --git a/flare/app/linreg_controller.py b/flare/app/linreg_controller.py index 03c216b..e4dc845 100644 --- a/flare/app/linreg_controller.py +++ b/flare/app/linreg_controller.py @@ -65,7 +65,7 @@ def control_flow(self, abort_signal: Signal, fl_ctx: FLContext): self.log_info(fl_ctx, f"{len(sites)}/{len(expected)} sites joined, n={scaling['n']}; features {scaling['features']}") beta = [0.0] * (len(scaling["features"]) + 1) - history, missing_rounds = [], {} + history, missing_rounds, round_contributions = [], {}, {} for r in range(1, self.rounds + 1): data = Shareable() data.update({"beta": beta, "mean": scaling["mean"], "std": scaling["std"], "lr": self.lr, @@ -73,7 +73,7 @@ def control_flow(self, abort_signal: Signal, fl_ctx: FLContext): replies = self._round("linreg_train", data, fl_ctx, abort_signal, targets=sites) if abort_signal.triggered: return - if len(replies) < self.min_clients: + if len(replies) < max(self.min_clients, 2): # one site's round would be its own local fit self.log_error(fl_ctx, f"round {r}: only {len(replies)} updates; stopping early") break miss = [s for s in sites if s not in replies] @@ -84,8 +84,14 @@ def control_flow(self, abort_signal: Signal, fl_ctx: FLContext): beta = new_beta history.append({"round": r, "sites": len(replies), "max_delta": delta, "coef": dict(zip(["intercept", *scaling["features"]], linreg.unstandardise(beta, scaling["mean"], scaling["std"])))}) + round_contributions[str(r)] = {s: rep["update"]["n"] for s, rep in replies.items()} self.log_info(fl_ctx, f"round {r}/{self.rounds}: {len(replies)} sites, max|Δβ|={delta:.2e}") + if not history: + # beta is still the zero start vector; recording it would release a non-fit + self.log_error(fl_ctx, "no FedAvg round completed; not recording a result") + return + coef = linreg.unstandardise(beta, scaling["mean"], scaling["std"]) merged = { "sites_expected": expected, "sites_reported": sites, "sites_missing": [s for s in expected if s not in sites], @@ -95,7 +101,7 @@ def control_flow(self, abort_signal: Signal, fl_ctx: FLContext): "stats": {"_linreg": {"ols": {"outcome": self.spec.outcome, "coef": dict(zip(["intercept", *scaling["features"]], coef))}, "n": scaling["n"], "history": history}}, - "contributions": {}, + "contributions": {"_rounds": round_contributions}, # who actually answered each round } check = disclosure_check.check(merged, self.spec.model_dump(), overseer_queue.release_log_path()) run_dir = overseer_queue.record(self.spec.model_dump(), merged, check) diff --git a/flare/app/logreg_controller.py b/flare/app/logreg_controller.py index 4f3a441..a1c1876 100644 --- a/flare/app/logreg_controller.py +++ b/flare/app/logreg_controller.py @@ -62,10 +62,9 @@ def control_flow(self, abort_signal: Signal, fl_ctx: FLContext): return sites = sorted(replies) - n_per_site = {s: replies[s]["n"] for s in sites} features = self.spec.variables beta = [0.0] * (len(features) + 1) - history, missing_rounds = [], {} + history, missing_rounds, round_contributions = [], {}, {} for r in range(1, self.rounds + 1): data = Shareable() @@ -76,7 +75,7 @@ def control_flow(self, abort_signal: Signal, fl_ctx: FLContext): miss = [s for s in sites if s not in replies] if miss: missing_rounds[str(r)] = miss # JSON object keys, as the result contract requires - if len(replies) < self.min_clients: + if len(replies) < max(self.min_clients, 2): # one site's round would be its own local fit self.log_error(fl_ctx, f"round {r}: only {len(replies)} updates; stopping early") break @@ -86,6 +85,7 @@ def control_flow(self, abort_signal: Signal, fl_ctx: FLContext): beta = new_beta history.append({"round": r, "sites": len(replies), "max_delta": delta, "coef": dict(zip(["intercept", *features], beta))}) + round_contributions[str(r)] = {s: rep["step"]["n"] for s, rep in replies.items()} self.log_info(fl_ctx, f"round {r}/{self.rounds}: {len(replies)} sites, max|Δβ|={delta:.2e}") if delta < self.tol: break @@ -95,15 +95,17 @@ def control_flow(self, abort_signal: Signal, fl_ctx: FLContext): self.log_error(fl_ctx, "no Newton round completed; not recording a result") return + # The fit's sample is the sites whose rows entered at least one round, not every site that joined. + fitted = {s: n for steps in round_contributions.values() for s, n in steps.items()} merged = { - "sites_expected": expected, "sites_reported": sites, "sites_missing": [s for s in expected if s not in sites], - "coverage": f"{len(sites)}/{len(expected)} sites", "n": sum(n_per_site.values()), "n_per_site": n_per_site, + "sites_expected": expected, "sites_reported": sorted(fitted), "sites_missing": [s for s in expected if s not in fitted], + "coverage": f"{len(fitted)}/{len(expected)} sites", "n": sum(fitted.values()), "n_per_site": fitted, "rejected_per_site": {}, "spec_hash": self.spec.spec_hash(), "sites_missing_rounds": missing_rounds, "method": {"mode": "newton_raphson", "rounds": len(history), "ridge": self.ridge, "tol": self.tol}, "stats": {"_logreg": {"logreg": {"outcome": self.spec.outcome, "coef": dict(zip(["intercept", *features], beta))}, - "n": sum(n_per_site.values()), "history": history}}, - "contributions": {}, + "n": sum(fitted.values()), "history": history}}, + "contributions": {"_rounds": round_contributions}, # who actually answered each round } check = disclosure_check.check(merged, self.spec.model_dump(), overseer_queue.release_log_path()) run_dir = overseer_queue.record(self.spec.model_dump(), merged, check) diff --git a/flare/app/logreg_executor.py b/flare/app/logreg_executor.py index 2a6a680..0f1bd55 100644 --- a/flare/app/logreg_executor.py +++ b/flare/app/logreg_executor.py @@ -6,8 +6,11 @@ logistic regression needs a fresh evaluation at each beta. logreg_step -> receive the current global beta, call the adapter's irls_step primitive (a fresh, local-only computation inside - the TRE at that beta), return (n, grad, hess). Each round is - audited; only p+1 numbers and a (p+1)x(p+1) matrix leave. + the TRE at that beta), return (n, grad, hess) if Safe Output + allows it (complete inputs, n >= max(k, p+2); see + safe_output.release_irls_step). Each round is audited, + accepted or rejected; only p+1 numbers and a (p+1)x(p+1) + matrix leave. """ from __future__ import annotations @@ -50,8 +53,7 @@ def execute(self, task_name: str, shareable: Shareable, fl_ctx: FLContext, abort outcome = self._adapter.local(self._spec.outcome) filters = self._adapter.local_filters(self._spec) step = self._adapter.irls_step(outcome, features, shareable["beta"], filters) - safe_output.audit_release(tre_id, self._spec, step["n"], {"_logreg": ["grad", "hess"]}, - "newton_raphson_round") + step = safe_output.release_irls_step(tre_id, self._spec, step, len(shareable["beta"])) reply = Shareable() reply["step"] = step return reply diff --git a/frontend/.env.example b/frontend/.env.example index 0fb612a..0e013a2 100644 --- a/frontend/.env.example +++ b/frontend/.env.example @@ -1 +1 @@ -NEXT_PUBLIC_API_BASE_URL=http://127.0.0.1:8080 +BACKEND_API_URL=http://127.0.0.1:8500 diff --git a/frontend/README.md b/frontend/README.md index ebbc012..7abe2f0 100644 --- a/frontend/README.md +++ b/frontend/README.md @@ -4,16 +4,16 @@ A Next.js interface for the Heimdall federated analysis API. It is intentionally ## Run -```powershell +```bash cd frontend npm install npm run dev ``` -The UI uses Webpack mode because the native Windows SWC binary may be unavailable. Browser requests use the same-origin `/api` proxy, so the browser does not need CORS access to the Python API. The proxy targets `http://127.0.0.1:8080` by default. +The UI uses Webpack mode because the native Windows SWC binary may be unavailable. Browser requests use the same-origin `/api` proxy, so the browser does not need CORS access to the Python API. The proxy targets `http://127.0.0.1:8500` by default. -```powershell -$env:BACKEND_API_URL = "http://127.0.0.1:8080" +```bash +export BACKEND_API_URL="http://127.0.0.1:8500" ``` The current backend API surface is: @@ -32,10 +32,10 @@ The current backend API surface is: -Start the Python API from the repository root on port 8080, then start this frontend: +Start the Python API from the repository root on port 8500, then start this frontend: -```powershell -python -m uvicorn server.api:app --host 127.0.0.1 --port 8080 --workers 1 +```bash +SERVER_OUT=server/api_out uvicorn server.api:app --port 8500 cd frontend npm run dev ``` diff --git a/frontend/app/api/[...path]/route.js b/frontend/app/api/[...path]/route.js index 66f4438..a32b676 100644 --- a/frontend/app/api/[...path]/route.js +++ b/frontend/app/api/[...path]/route.js @@ -1,4 +1,4 @@ -const backendBase = process.env.BACKEND_API_URL || process.env.NEXT_PUBLIC_API_BASE_URL || 'http://127.0.0.1:8080' +const backendBase = process.env.BACKEND_API_URL || process.env.NEXT_PUBLIC_API_BASE_URL || 'http://127.0.0.1:8500' async function forward(request, context) { const { path = [] } = await context.params diff --git a/frontend/app/page.jsx b/frontend/app/page.jsx index 5cd9624..ecc6958 100644 --- a/frontend/app/page.jsx +++ b/frontend/app/page.jsx @@ -197,10 +197,13 @@ export default function Home() { } async function decide(item, decision) { + const specHash = item.spec_hash || item.id try { - await api(`/overseer/${item.spec_hash || item.id}/${decision}`, { method: 'POST', body: JSON.stringify({ note: '', by: 'researcher' }) }) - setPending((items) => items.filter((entry) => (entry.spec_hash || entry.id) !== (item.spec_hash || item.id))) + await api(`/overseer/${specHash}/${decision}`, { method: 'POST', body: JSON.stringify({ note: '', by: 'researcher' }) }) + setPending((items) => items.filter((entry) => (entry.spec_hash || entry.id) !== specHash)) setNotice(`Result ${decision}d.`) + // the displayed run stopped polling when it was flagged; show the decision (and any released result) + if (runId && run?.spec_hash === specHash) setRun(await api(`/run/${runId}`)) } catch (error) { setNotice(error.message) } diff --git a/frontend/package-lock.json b/frontend/package-lock.json index 9a43375..e5583be 100644 --- a/frontend/package-lock.json +++ b/frontend/package-lock.json @@ -8,10 +8,10 @@ "name": "heimdall", "version": "0.1.0", "dependencies": { - "lucide-react": "latest", - "next": "latest", - "react": "latest", - "react-dom": "latest" + "lucide-react": "1.47.0", + "next": "16.3.5", + "react": "19.3.0", + "react-dom": "19.3.0" } }, "node_modules/@emnapi/runtime": { @@ -338,6 +338,7 @@ "resolved": "https://registry.npmjs.org/react/-/react-19.3.0.tgz", "integrity": "sha512-E8LUcbtBWt20bbl2YoHfx4ZDBdxVTfOKtCZn9cDSJ4l6/nuoApcpIBcj47t2wZoVX8g2ZHuMHbiShgCR1T5Sog==", "license": "MIT", + "peer": true, "engines": { "node": ">=0.10.0" } @@ -347,6 +348,7 @@ "resolved": "https://registry.npmjs.org/react-dom/-/react-dom-19.3.0.tgz", "integrity": "sha512-JDk8dgif51OjFoDE70+OT9ICyYr+69HlmihNwp1+Nsfbna3t5sIiCa9ZJktDmQ4/1b/rn26hIAR2uYXDMr5r0Q==", "license": "MIT", + "peer": true, "dependencies": { "scheduler": "^0.28.0" }, diff --git a/frontend/package.json b/frontend/package.json index 5188ff9..64c1d29 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -10,9 +10,9 @@ "test": "node --import ./register-jsx-loader.mjs --test" }, "dependencies": { - "next": "latest", - "react": "latest", - "react-dom": "latest", - "lucide-react": "latest" + "next": "16.3.5", + "react": "19.3.0", + "react-dom": "19.3.0", + "lucide-react": "1.47.0" } } diff --git a/projects.yaml b/projects.yaml index 48db3df..446da12 100644 --- a/projects.yaml +++ b/projects.yaml @@ -1,10 +1,14 @@ # Safe Projects allow-list. An AnalysisSpec whose project_id is not listed here # is rejected by every adapter before anything is computed (and audited as such). # `sites: all` or a list of tre_ids that have approved the project. +# `min_cell_size` is the project's disclosure floor: a request may ask for a +# larger k, never a smaller one (default 5 when a project omits it). projects: ncfh-2026-demo: description: NCFH 2026 hackathon demo -- allele frequency + linear regression sites: all + min_cell_size: 5 hackathon-test: description: Adapter development and CI sites: all + min_cell_size: 5 diff --git a/scripts/limit_server_jobs.py b/scripts/limit_server_jobs.py new file mode 100644 index 0000000..04b09b0 --- /dev/null +++ b/scripts/limit_server_jobs.py @@ -0,0 +1,70 @@ +#!/usr/bin/env python +"""Cap the provisioned FLARE server's job scheduler at 1 concurrent job. + +NVFlare's provisioned server kit ships local/resources.json.default with +DefaultJobScheduler's max_jobs=4 (nvflare/lighter/templates/master_template.yml, +local_server_resources). At runtime NVFlare uses local/resources.json if it +exists, otherwise falls back to local/resources.json.default -- it does not +merge the two (nvflare.apis.workspace.Workspace._fallback_path). The server's +overseer queue and release log (server/overseer_queue.py) are not safe for +concurrent writer processes, so the federation must run one job at a time. + +This writes local/resources.json next to the server kit's .default file, a +copy of the complete default document with components[job_scheduler].args +.max_jobs forced to 1. The .default file is left untouched. Idempotent: reruns +overwrite resources.json with the same result as long as the .default is +unchanged. + + python scripts/limit_server_jobs.py flare/workspace//prod_00 +""" +from __future__ import annotations + +import argparse +import json +from pathlib import Path + +DEFAULT_NAME = "resources.json.default" +TARGET_NAME = "resources.json" + + +def find_server_kit_default(prod_dir: Path) -> Path: + """Return the single */local/resources.json.default under prod_dir whose + components include a job_scheduler; fail loudly if that isn't exactly one.""" + matches = [] + for default_path in sorted(prod_dir.glob(f"*/local/{DEFAULT_NAME}")): + doc = json.loads(default_path.read_text()) + if any(c.get("id") == "job_scheduler" for c in doc.get("components", [])): + matches.append(default_path) + if len(matches) != 1: + raise SystemExit( + f"expected exactly one server kit ({DEFAULT_NAME} with a job_scheduler component) " + f"under {prod_dir}, found {len(matches)}: {matches}" + ) + return matches[0] + + +def limit_max_jobs(default_path: Path, max_jobs: int = 1) -> Path: + """Write local/resources.json next to default_path: the full default document + with components[job_scheduler].args.max_jobs set to max_jobs.""" + doc = json.loads(default_path.read_text()) + schedulers = [c for c in doc.get("components", []) if c.get("id") == "job_scheduler"] + if len(schedulers) != 1: + raise SystemExit(f"expected exactly one job_scheduler component in {default_path}, found {len(schedulers)}") + schedulers[0]["args"]["max_jobs"] = max_jobs + target = default_path.with_name(TARGET_NAME) + target.write_text(json.dumps(doc, indent=2) + "\n") + return target + + +def main() -> None: + ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + ap.add_argument("prod_dir", type=Path, help="provisioned prod_XX dir, e.g. flare/workspace//prod_00") + ap.add_argument("--max-jobs", type=int, default=1) + a = ap.parse_args() + default_path = find_server_kit_default(a.prod_dir) + target = limit_max_jobs(default_path, a.max_jobs) + print(f"wrote {target} (max_jobs={a.max_jobs})") + + +if __name__ == "__main__": + main() diff --git a/scripts/local_federation.sh b/scripts/local_federation.sh index 4aac73a..3644221 100755 --- a/scripts/local_federation.sh +++ b/scripts/local_federation.sh @@ -28,6 +28,7 @@ case "${1:-}" in $PY scripts/gen_sites.py > /dev/null rm -rf $FED/workspace/$PROJECT/prod_* $PY -m nvflare.lighter.provision -p flare/project.yml -w $FED/workspace --force > /dev/null + $PY scripts/limit_server_jobs.py "$KITS" > /dev/null git checkout -q docker-compose.yml flare/project.yml 2>/dev/null || true # generated files stay as in sites.yaml nohup $PY scripts/dev_tres.py --port 8100 > $FED/tres.log 2>&1 & echo $! > $FED/tres.pid diff --git a/scripts/provision.sh b/scripts/provision.sh index da5e829..46e0740 100755 --- a/scripts/provision.sh +++ b/scripts/provision.sh @@ -9,4 +9,5 @@ PROJECT=$($PY -c "import yaml; print(yaml.safe_load(open('flare/project.yml'))[' $PY scripts/gen_sites.py >/dev/null rm -rf "flare/workspace/$PROJECT"/prod_* $PY -m nvflare.lighter.provision -p flare/project.yml -w flare/workspace --force >/dev/null +$PY scripts/limit_server_jobs.py "flare/workspace/$PROJECT/prod_00" >/dev/null echo "flare/workspace/$PROJECT/prod_00" diff --git a/server/aggregate.py b/server/aggregate.py index f57f5bc..8d7449b 100644 --- a/server/aggregate.py +++ b/server/aggregate.py @@ -26,7 +26,9 @@ def combine(results: Iterable[AggregateResult], expected_sites: list[str] | None "rejected_per_site": {r.tre_id: r.rejected for r in results if r.rejected}, "spec_hash": next((r.spec_hash for r in results if r.spec_hash), ""), "stats": {}, - "contributions": {}, # var -> cell -> {site: count}; used by the dominance check, not released + # var -> cell -> {site: count}; used by the dominance check, never released. Cells are + # genotype levels, "count" (observed values of a scalar variable) or "n" (regression sample) + "contributions": {}, } variables = sorted({v for r in results for v in r.stats}) for v in variables: @@ -49,6 +51,7 @@ def combine(results: Iterable[AggregateResult], expected_sites: list[str] | None agg.update(count=int(n), sum=s, sum_sq=ss, mean=s / n if n else None, var=(ss - s * s / n) / (n - 1) if n > 1 else None, min=min(p["min"] for p in parts.values()), max=max(p["max"] for p in parts.values())) + out["contributions"].setdefault(v, {})["count"] = {site: int(p["count"]) for site, p in parts.items()} if parts and all("gram" in p for p in parts.values()): G = sum(np.array(p["gram"]["matrix"]) for p in parts.values()) cols = next(iter(parts.values()))["gram"]["cols"] @@ -59,5 +62,6 @@ def combine(results: Iterable[AggregateResult], expected_sites: list[str] | None except np.linalg.LinAlgError as e: agg["ols"] = {"error": str(e)} agg["n"] = int(sum(p["gram"]["n"] for p in parts.values())) + out["contributions"].setdefault(v, {})["n"] = {site: int(p["gram"]["n"]) for site, p in parts.items()} out["stats"][v] = agg return out diff --git a/server/analysis_service.py b/server/analysis_service.py index 76b382b..11b28ed 100644 --- a/server/analysis_service.py +++ b/server/analysis_service.py @@ -4,20 +4,19 @@ may instead run adapters inside TREs and use outbound FLARE communication. Run one API worker with exclusive ownership of SERVER_OUT; the existing JSON queue is not safe for simultaneous writers in other processes (including the -overseer CLI). The lock below serializes this process's check/record operations. +overseer CLI). overseer_queue.RELEASE_LOCK serializes this process's +check/record operations and overseer decisions. """ from __future__ import annotations from concurrent.futures import ThreadPoolExecutor -import json import logging import math import os -from threading import Lock import httpx -from adapters import registry +from adapters import registry, safe_output from adapters.base import AggregateResult, VariableStats from flare.app import linreg, logreg from scripts.sites import load_sites @@ -26,7 +25,6 @@ from spec.analysis_spec import AnalysisSpec LOG = logging.getLogger(__name__) -_RELEASE_LOCK = Lock() def _validate_result(result: AggregateResult, tre_id: str, spec: AnalysisSpec) -> None: @@ -79,30 +77,11 @@ def _run_site(tre_id: str, spec: AnalysisSpec, timeout: float): LOG.warning("Could not close TRE %s client", tre_id) -def _validate_irls_step(step, p: int) -> None: - """Same spirit as _validate_result's Gram check, for one round's worth of - logistic-regression sufficient statistics from one site.""" - def count(value): - return type(value) is int and value >= 0 - - if not isinstance(step, dict) or not count(step.get("n")): - raise ValueError("invalid irls_step result") - json.dumps(step, allow_nan=False) - grad, hess = step.get("grad"), step.get("hess") - if (not isinstance(grad, list) or len(grad) != p - or any(type(v) not in (int, float) or not math.isfinite(v) for v in grad)): - raise ValueError("invalid irls_step gradient") - if (not isinstance(hess, list) or len(hess) != p - or any(not isinstance(row, list) or len(row) != p - or any(type(v) not in (int, float) or not math.isfinite(v) for v in row) - for row in hess)): - raise ValueError("invalid irls_step Hessian") - - def _run_logreg_step(tre_id: str, spec: AnalysisSpec, beta: list[float], timeout: float): """One Newton-Raphson round's worth of local computation at one site. Mirrors _run_site's shape/error handling; called once per round, not once - per request, since irls_step must be re-evaluated at the current beta.""" + per request, since irls_step must be re-evaluated at the current beta. + The round leaves through the same Safe Output step as the FLARE executor's.""" adapter = None try: adapter = registry.load(tre_id) @@ -111,8 +90,9 @@ def _run_logreg_step(tre_id: str, spec: AnalysisSpec, beta: list[float], timeout outcome = adapter.local(spec.outcome) filters = adapter.local_filters(spec) step = adapter.irls_step(outcome, features, beta, filters) - _validate_irls_step(step, len(beta)) - return step, None + return safe_output.release_irls_step(tre_id, spec, step, len(beta)), None + except safe_output.Rejected: + return None, "rejected" except httpx.TimeoutException: return None, "timeout" except httpx.RequestError: @@ -132,6 +112,18 @@ def _valid_statistics(merged: dict, spec: AnalysisSpec) -> bool: if spec.analysis_type == "allele_freq": value = merged["stats"].get(spec.variables[0], {}).get("allele_freq") return isinstance(value, (int, float)) and math.isfinite(value) + if spec.analysis_type == "fed_stats": + # every variable carries observed values, and nothing released is non-finite + def observed(s: dict) -> bool: + if s.get("genotype_counts"): + return sum(s["genotype_counts"].values()) > 0 + return (s.get("count") or 0) > 0 + + released = [merged["stats"].get(v) or {} for v in spec.variables] + scalars = [x for s in released for key, x in s.items() + if key in ("count", "sum", "sum_sq", "mean", "var", "min", "max") and x is not None] + return (all(observed(s) for s in released) + and all(isinstance(x, (int, float)) and math.isfinite(x) for x in scalars)) ols = merged["stats"].get("_linreg", {}).get("ols", {}) coef = ols.get("coef", {}) return bool(coef) and "error" not in ols and all(math.isfinite(v) for v in coef.values()) @@ -187,7 +179,7 @@ def analyse(spec: AnalysisSpec, fedavg_rounds: int = 0) -> tuple[int, dict]: merged["stats"]["_linreg"] = _fedavg(results, spec, fedavg_rounds) merged["method"] = {"mode": "fedavg", "rounds": fedavg_rounds, "local_steps": 1, "lr": 1.0} spec_dict = spec.model_dump() - with _RELEASE_LOCK: + with overseer_queue.RELEASE_LOCK: check = disclosure_check.check(merged, spec_dict, overseer_queue.release_log_path()) # Disclosure approval alone does not mean an OLS solve was successful. if results and check["decision"] == "OK" and not _valid_statistics(merged, spec): @@ -239,12 +231,11 @@ def analyse_logreg(spec: AnalysisSpec, rounds: int = 25, tol: float = 1e-8, ridg else: gated.append(result) - merged = combine(gated, expected) - merged["spec_hash"] = spec.spec_hash() spec_dict = spec.model_dump() - sites = sorted(r.tre_id for r in gated) history: list[dict] = [] + rounds_seen: dict[str, dict[str, int]] = {} # round -> {site: n} of the steps it accepted + fit = None if len(sites) >= 2: # matches disclosure_check's min_sites floor; no point running rounds otherwise beta = [0.0] * (len(spec.variables) + 1) for r in range(1, rounds + 1): @@ -267,15 +258,23 @@ def analyse_logreg(spec: AnalysisSpec, rounds: int = 25, tol: float = 1e-8, ridg beta = new_beta history.append({"round": r, "sites": len(round_steps), "max_delta": delta, "coef": dict(zip(["intercept", *spec.variables], beta))}) + rounds_seen[str(r)] = {t: s["n"] for t, s in round_steps.items()} if delta < tol: break - merged["stats"] = {"_logreg": {"logreg": {"outcome": spec.outcome, - "coef": dict(zip(["intercept", *spec.variables], beta))}, - "n": merged["n"], "history": history}} + fit = {"outcome": spec.outcome, "coef": dict(zip(["intercept", *spec.variables], beta))} + + # The fit's sample is the sites whose rows entered at least one Newton round -- not every + # site that passed the gate (one rejected in every round contributed nothing). + contributed = {t for steps in rounds_seen.values() for t in steps} + merged = combine([g for g in gated if g.tre_id in contributed], expected) + merged["spec_hash"] = spec.spec_hash() + merged["contributions"]["_rounds"] = rounds_seen + if fit: + merged["stats"] = {"_logreg": {"logreg": fit, "n": merged["n"], "history": history}} merged["method"] = {"mode": "newton_raphson", "rounds": len(history), "ridge": ridge, "tol": tol} merged["sites_failed"] = failed - with _RELEASE_LOCK: + with overseer_queue.RELEASE_LOCK: check = disclosure_check.check(merged, spec_dict, overseer_queue.release_log_path()) if gated and check["decision"] == "OK" and not _valid_logreg_statistics(merged, spec): check = {**check, "decision": "FLAGGED", diff --git a/server/api.py b/server/api.py index af17c0b..b2275b9 100644 --- a/server/api.py +++ b/server/api.py @@ -192,6 +192,26 @@ def example(name: str): _RUNS_LOCK = threading.Lock() +def _decision_update(approve: bool, result: dict | None) -> dict: + """How a flagged run reports an overseer's decision.""" + return ({"status": "completed", "decision": "APPROVED", "result": result} if approve + else {"status": "rejected", "decision": "REJECTED"}) + + +def _decided_meanwhile(spec_hash: str) -> dict: + """An overseer can decide between record() queuing this run's result and the run's + status being written. Read under RELEASE_LOCK, the queue and release log say whether + one did: this run queued the item, so if it is gone, a later log line removed it.""" + if any(i["spec_hash"] == spec_hash for i in overseer_queue._load_queue()): + return {} + last = overseer_queue.latest_log_entry(spec_hash) + if not last or last["check"] != "FLAGGED" or last["decision"] not in ("RELEASED", "REJECTED"): + return {} # e.g. superseded by a later, auto-released run of the same spec + approve = last["decision"] == "RELEASED" + released = overseer_queue.out_dir() / spec_hash / "released.json" + return _decision_update(approve, json.loads(released.read_text()) if approve else None) + + def _run_in_background(run_id: str, spec: AnalysisSpec, fedavg_rounds: int) -> None: with _RUNS_LOCK: _RUNS[run_id]["status"] = "running" @@ -200,7 +220,9 @@ def _run_in_background(run_id: str, spec: AnalysisSpec, fedavg_rounds: int) -> N except Exception: LOG.exception("Run %s failed", run_id) body = {"status": "failed", "error": "analysis_failed", "spec_hash": spec.spec_hash()} - with _RUNS_LOCK: + with overseer_queue.RELEASE_LOCK, _RUNS_LOCK: # no decision lands between this check and the write + if body.get("status") == "flagged": + body = {**body, **_decided_meanwhile(body["spec_hash"])} _RUNS[run_id].update(body, finished=_now()) @@ -235,12 +257,22 @@ def overseer_decide(spec_hash: str, decision: str, body: OverseerDecision | None if decision not in ("approve", "reject"): raise HTTPException(404, "decision must be approve or reject") body = body or OverseerDecision() - try: - d = overseer_queue.decide(spec_hash, decision == "approve", body.note, body.by) - except KeyError: - raise HTTPException(404, "not in queue") - return {"spec_hash": spec_hash, "decision": "RELEASED" if decision == "approve" else "REJECTED", - "released": (d / "released.json").exists()} + approve = decision == "approve" + # The decision, reading what it released, and every waiting run's status change as one + # step: no rerun or finishing background run interleaves (see _decided_meanwhile). + with overseer_queue.RELEASE_LOCK: + try: + d = overseer_queue.decide(spec_hash, approve, body.note, body.by) + except KeyError: + raise HTTPException(404, "not in queue") + update = _decision_update(approve, json.loads((d / "released.json").read_text()) if approve else None) + # Runs of this spec still waiting on the overseer now report the decision (the queue + # holds the latest run's result, which is what an approval releases). + with _RUNS_LOCK: + for run in _RUNS.values(): + if run.get("spec_hash") == spec_hash and run["status"] == "flagged": + run.update(update) + return {"spec_hash": spec_hash, "decision": "RELEASED" if approve else "REJECTED", "released": approve} @app.get("/audit", response_model=AuditLog) diff --git a/server/disclosure_check.py b/server/disclosure_check.py index f27f743..78ef985 100644 --- a/server/disclosure_check.py +++ b/server/disclosure_check.py @@ -1,9 +1,13 @@ """Server-side disclosure check on the MERGED result (N-aware). - k-anonymity every released count cell in the merged table >= k - dominance no single site contributes more than `dominance` of a cell - (a near-single-contributor cell is effectively that site's own output) - min sites a federated release must combine >= min_sites sites + k-anonymity every released count cell in the merged table >= k, where k is + the request's min_cell_size raised to the project's floor + dominance no single site contributes more than `dominance` of a cell, of a + variable's observed values, of a regression's sample, or of any + released federated round (a near-single-contributor release is + effectively that site's own output) + min sites a federated release must combine >= min_sites sites, and so must + every released round of an iterative fit site suppression any site's Safe Output filter suppressed something: the merged table is incomplete and a human should see why differencing a previous release with the same analysis + variables whose n @@ -18,14 +22,26 @@ import json from pathlib import Path +from adapters.safe_output import effective_min_cell_size + def _signature(spec: dict) -> str: vars_ = sorted(spec.get("variables", []) + ([spec["outcome"]] if spec.get("outcome") else [])) return f"{spec['analysis_type']}|{','.join(vars_)}" +def _dominant(contrib: dict[str, int], dominance: float) -> tuple[str, float] | None: + """(site, share) when one of several contributing sites holds more than `dominance`.""" + total = sum(contrib.values()) + if len(contrib) < 2 or not total: + return None + site, top = max(contrib.items(), key=lambda kv: kv[1]) + return (site, top / total) if top / total > dominance else None + + def check(merged: dict, spec: dict, release_log: Path | None = None, *, dominance: float = 0.9, min_sites: int = 2) -> dict: - k = int(spec.get("min_cell_size", 5)) + k = effective_min_cell_size(spec.get("project_id"), int(spec.get("min_cell_size", 5))) + contributions = merged.get("contributions", {}) reasons: list[str] = [] if len(merged["sites_reported"]) < min_sites: @@ -39,13 +55,23 @@ def check(merged: dict, spec: dict, release_log: Path | None = None, *, dominanc for lvl, c in (st.get(table) or {}).items(): if c < k: reasons.append(f"k_anon:{var}.{table}.{lvl}={c}<{k}") - contrib = merged.get("contributions", {}).get(var, {}).get(lvl, {}) - if c and contrib: - top_site, top = max(contrib.items(), key=lambda kv: kv[1]) - if len(contrib) > 1 and top / c > dominance: - reasons.append(f"dominance:{var}.{table}.{lvl}:{top_site}={top / c:.2f}>{dominance}") + if hit := _dominant(contributions.get(var, {}).get(lvl, {}), dominance): + reasons.append(f"dominance:{var}.{table}.{lvl}:{hit[0]}={hit[1]:.2f}>{dominance}") if st.get("count") is not None and st["count"] < k: reasons.append(f"k_anon:{var}.count={st['count']}<{k}") + for cell in ("count", "n"): # a variable's observed values; a regression's sample + if hit := _dominant(contributions.get(var, {}).get(cell, {}), dominance): + reasons.append(f"dominance:{var}.{cell}:{hit[0]}={hit[1]:.2f}>{dominance}") + + # every round's coefficients are released (history), so every round is checked + rounds = contributions.get("_rounds", {}) + if thin := [(r, len(contrib)) for r, contrib in rounds.items() if len(contrib) < min_sites]: + r, count = min(thin, key=lambda x: x[1]) + reasons.append(f"min_sites:round={r}:{count}<{min_sites}") + per_round = [(r, *hit) for r, contrib in rounds.items() if (hit := _dominant(contrib, dominance))] + if per_round: + r, site, share = max(per_round, key=lambda x: x[2]) + reasons.append(f"dominance:round={r}:{site}={share:.2f}>{dominance}") if release_log and release_log.exists(): sig, n = _signature(spec), merged["n"] diff --git a/server/overseer_queue.py b/server/overseer_queue.py index 2521dbf..8c45ed3 100644 --- a/server/overseer_queue.py +++ b/server/overseer_queue.py @@ -8,21 +8,30 @@ Files (under SERVER_OUT, default server/out/): overseer_queue.json pending items release_log.jsonl every release decision (auto-OK, approved, rejected), keyed by spec hash - /result.json merged result (written for every run) - /released.json only after OK / approval -- this is what leaves the server + /result.json merged result of the latest run of that spec + /released.json only while the latest run is OK / approved -- what left the server + +One process owns a SERVER_OUT: RELEASE_LOCK serialises this process's check -> +record and decisions, and queue writes are atomic, but nothing coordinates two +processes (API, FLARE server, this CLI) writing the same directory. """ from __future__ import annotations import argparse import json import os +import secrets import sys +import threading import time from pathlib import Path from server.schemas import MergedResult, ReleasedResult ROOT = Path(__file__).resolve().parent.parent +# Held across disclosure check -> record (the check reads the release log that record +# and decide append to) and around every decision. Re-entrant: record/decide take it too. +RELEASE_LOCK = threading.RLock() def out_dir() -> Path: @@ -44,8 +53,21 @@ def _load_queue() -> list[dict]: return json.loads(p.read_text()) if p.exists() else [] +def _write_atomic(path: Path, text: str) -> None: + """Readers see the old file or the new one, never a partial write. The temp file is + opened normally (not mkstemp's 0600) so the result keeps the usual umask permissions.""" + tmp = path.with_name(f".{path.name}.{secrets.token_hex(8)}.tmp") + try: + with open(tmp, "x") as f: + f.write(text) + os.replace(tmp, path) + except BaseException: + tmp.unlink(missing_ok=True) + raise + + def _save_queue(q: list[dict]) -> None: - queue_path().write_text(json.dumps(q, indent=2)) + _write_atomic(queue_path(), json.dumps(q, indent=2)) def _now() -> str: @@ -53,8 +75,19 @@ def _now() -> str: def _releasable(merged: dict) -> dict: - """What actually leaves: the ReleasedResult projection (no per-site contributions).""" - return ReleasedResult.model_validate({k: v for k, v in merged.items() if k != "contributions"}).model_dump(exclude_none=True) + """What actually leaves: the ReleasedResult projection -- no per-site contributions, and each + per-site suppression named by its location only (the reason detail holds the small count).""" + released = {k: v for k, v in merged.items() if k != "contributions"} + released["rejected_per_site"] = {site: [reason.split(":", 1)[0] for reason in reasons] + for site, reasons in merged.get("rejected_per_site", {}).items()} + return ReleasedResult.model_validate(released).model_dump(exclude_none=True) + + +def latest_log_entry(spec_hash: str) -> dict | None: + """The newest release-log line for a spec hash (QUEUED / RELEASED / REJECTED), if any.""" + p = release_log_path() + lines = p.read_text().splitlines() if p.exists() else [] + return next((e for e in map(json.loads, reversed(lines)) if e.get("spec_hash") == spec_hash), None) def log_release(spec: dict, merged: dict, check: dict, decision: str, note: str = "", by: str = "auto") -> None: @@ -70,37 +103,44 @@ def log_release(spec: dict, merged: dict, check: dict, decision: str, note: str def record(spec: dict, merged: dict, check: dict) -> Path: """Called by the FLARE controller / HTTP API after the disclosure check. Returns the run dir. - The merged result must satisfy the MergedResult contract; a shape drift fails here, loudly.""" + The merged result must satisfy the MergedResult contract; a shape drift fails here, loudly. + The run dir and queue describe this (latest) run of the spec hash: an earlier run's + release or pending queue entry does not survive it.""" merged = MergedResult.model_validate(merged).model_dump(exclude_none=True) d = out_dir() / (merged.get("spec_hash") or "unknown") - d.mkdir(parents=True, exist_ok=True) - (d / "spec.json").write_text(json.dumps(spec, indent=2)) - (d / "result.json").write_text(json.dumps(merged, indent=2)) - (d / "check.json").write_text(json.dumps(check, indent=2)) - if check["decision"] == "OK": - (d / "released.json").write_text(json.dumps(_releasable(merged), indent=2)) - log_release(spec, merged, check, "RELEASED") - else: + with RELEASE_LOCK: + d.mkdir(parents=True, exist_ok=True) + (d / "spec.json").write_text(json.dumps(spec, indent=2)) + (d / "result.json").write_text(json.dumps(merged, indent=2)) + (d / "check.json").write_text(json.dumps(check, indent=2)) q = [i for i in _load_queue() if i["spec_hash"] != merged.get("spec_hash")] - q.append({"spec_hash": merged.get("spec_hash"), "queued": _now(), "project_id": spec.get("project_id"), - "analysis_type": spec.get("analysis_type"), "reasons": check["reasons"], "dir": str(d)}) + if check["decision"] == "OK": + _write_atomic(d / "released.json", json.dumps(_releasable(merged), indent=2)) + log_release(spec, merged, check, "RELEASED") + else: + (d / "released.json").unlink(missing_ok=True) + q.append({"spec_hash": merged.get("spec_hash"), "queued": _now(), "project_id": spec.get("project_id"), + "analysis_type": spec.get("analysis_type"), "reasons": check["reasons"], "dir": str(d)}) + log_release(spec, merged, check, "QUEUED") _save_queue(q) - log_release(spec, merged, check, "QUEUED") return d def decide(spec_hash: str, approve: bool, note: str, by: str) -> Path: """Approve or reject a queued result. Raises KeyError if it is not queued.""" - q = _load_queue() - item = next((i for i in q if i["spec_hash"] == spec_hash), None) - if item is None: - raise KeyError(f"{spec_hash}: not in queue") - d = Path(item["dir"]) - spec, merged, check = (json.loads((d / f).read_text()) for f in ("spec.json", "result.json", "check.json")) - if approve: - (d / "released.json").write_text(json.dumps(_releasable(merged), indent=2)) - log_release(spec, merged, check, "RELEASED" if approve else "REJECTED", note, by) - _save_queue([i for i in q if i["spec_hash"] != spec_hash]) + with RELEASE_LOCK: + q = _load_queue() + item = next((i for i in q if i["spec_hash"] == spec_hash), None) + if item is None: + raise KeyError(f"{spec_hash}: not in queue") + d = Path(item["dir"]) + spec, merged, check = (json.loads((d / f).read_text()) for f in ("spec.json", "result.json", "check.json")) + if approve: + _write_atomic(d / "released.json", json.dumps(_releasable(merged), indent=2)) + else: + (d / "released.json").unlink(missing_ok=True) + log_release(spec, merged, check, "RELEASED" if approve else "REJECTED", note, by) + _save_queue([i for i in q if i["spec_hash"] != spec_hash]) return d diff --git a/server/schemas.py b/server/schemas.py index 32fdb3d..c293cfe 100644 --- a/server/schemas.py +++ b/server/schemas.py @@ -139,11 +139,12 @@ class RunRequest(_Strict): class RunStatus(_Strict): run_id: str - status: Literal["queued", "running", "completed", "flagged", "failed"] + # a flagged run becomes completed (APPROVED, with its result) or rejected once an overseer decides + status: Literal["queued", "running", "completed", "flagged", "rejected", "failed"] submitted: str finished: str | None = None spec_hash: str | None = None - decision: Literal["OK", "FLAGGED"] | None = None + decision: Literal["OK", "FLAGGED", "APPROVED", "REJECTED"] | None = None reasons: list[PublicReason] = Field(default_factory=list) sites_expected: list[str] = Field(default_factory=list) sites_reported: list[str] = Field(default_factory=list) diff --git a/tests/test_analysis_api.py b/tests/test_analysis_api.py index 3c38712..67f1e5c 100644 --- a/tests/test_analysis_api.py +++ b/tests/test_analysis_api.py @@ -140,7 +140,8 @@ def test_logistic_regression_site_down_for_whole_request(api, sites): def test_logistic_regression_site_fails_mid_rounds(api, sites): """A site that passes the gate but then goes dark partway through Newton's method should not corrupt the run: it's dropped from later rounds' sums and - reported in sites_failed, and the fit still converges close to ground truth + reported in sites_failed (and, having contributed no round, it is not among the + fit's sites), and the fit still converges close to ground truth using the sites that kept answering.""" client, failed, called = api bad = sites[-1]["tre_id"] @@ -164,7 +165,7 @@ def load(tre_id): assert response.status_code == 200 body = assert_safe(response, True) assert body["decision"] == "OK" - assert bad in body["sites_reported"] # it did gate in successfully + assert bad not in body["sites_reported"] # gated in, but no round used its rows assert body["sites_failed"] == {bad: "unavailable"} coef = body["result"]["stats"]["_logreg"]["logreg"]["coef"] assert max(abs(coef[k] - GT["logreg"]["coef"][k]) for k in coef) < 0.5 # close, not exact: 2/3 sites @@ -195,7 +196,7 @@ def load(tre_id): monkeypatch.undo() assert response.status_code == 200 body = assert_safe(response) - assert body["decision"] == "FLAGGED" and body["reasons"] == ["analysis"] + assert body["decision"] == "FLAGGED" and body["reasons"] == ["min_sites"] # no site contributed a round assert body["sites_failed"] == dict.fromkeys(bad, "unavailable") @@ -406,3 +407,63 @@ def fail(*args, **kwargs): assert client.get("/health").json() == {"status": "ok"} ''' subprocess.run([sys.executable, "-c", script], cwd=ROOT, check=True) + + +def test_logistic_regression_audits_every_round_at_every_site(api, sites): + """The direct API releases each Newton round's gradient/Hessian through the same + Safe Output step the FLARE executor uses: one audit line per site per round.""" + import os + from pathlib import Path + + client = api[0] + body = client.post("/logistic-regression", json={ + "features": GT["logreg"]["features"], "target": GT["logreg"]["outcome"], "project_id": PROJECT}).json() + assert body["decision"] == "OK" + rounds = body["result"]["method"]["rounds"] + audit = [json.loads(line) for line in (Path(os.environ["AUDIT_DIR"]) / "audit.jsonl").read_text().splitlines()] + per_round = [e for e in audit if e["spec_hash"] == body["spec_hash"] and e["decision"] == "OK:newton_raphson_round"] + assert sorted(e["tre_id"] for e in per_round) == sorted(s["tre_id"] for s in sites for _ in range(rounds)) + run = json.loads((overseer_queue.out_dir() / body["spec_hash"] / "result.json").read_text()) + assert run["contributions"]["_rounds"] == {str(r): GT["n_per_site"] for r in range(1, rounds + 1)} + + +def test_logistic_regression_fit_counts_only_contributing_sites(api, sites): + """A site that passes the gate but is rejected by Safe Output in every round put no + rows into the fit: it is not among the fit's sites and not in its n.""" + client = api[0] + bad = sites[-1]["tre_id"] + original = registry.load + + def load(tre_id): + adapter = original(tre_id) + if tre_id == bad: + adapter.irls_step = lambda *args, **kwargs: {"n": GT["n_per_site"][bad], "complete": False, + "grad": None, "hess": None} + return adapter + + monkeypatch = pytest.MonkeyPatch() + monkeypatch.setattr(registry, "load", load) + try: + body = client.post("/logistic-regression", json={ + "features": GT["logreg"]["features"], "target": GT["logreg"]["outcome"], "project_id": PROJECT}).json() + finally: + monkeypatch.undo() + kept = sorted(s["tre_id"] for s in sites if s["tre_id"] != bad) + assert body["decision"] == "OK" and body["sites_failed"] == {bad: "rejected"} + assert sorted(body["sites_reported"]) == kept + result = body["result"] + assert sorted(result["n_per_site"]) == kept + assert result["n"] == result["stats"]["_logreg"]["n"] == sum(GT["n_per_site"][t] for t in kept) + + +def test_fed_stats_is_valid_only_with_observed_values(): + from spec.analysis_spec import AnalysisSpec + + genotype = AnalysisSpec(analysis_type="fed_stats", variables=["snp_rs001"], project_id=PROJECT) + empty = {"genotype_counts": {}, "allele_counts": {"minor": 0, "major": 0, "n_alleles": 0}} + assert not analysis_service._valid_statistics({"stats": {"snp_rs001": empty}}, genotype) + assert analysis_service._valid_statistics({"stats": {"snp_rs001": {"genotype_counts": {"0": 50, "1": 9}}}}, genotype) + age = AnalysisSpec(analysis_type="fed_stats", variables=["age", "bmi"], project_id=PROJECT) + observed = {"count": 10, "sum": 500.0, "sum_sq": 25100.0, "mean": 50.0, "var": 11.1, "min": 40.0, "max": 60.0} + assert not analysis_service._valid_statistics({"stats": {"age": observed, "bmi": {"count": 0}}}, age) + assert analysis_service._valid_statistics({"stats": {"age": observed, "bmi": observed}}, age) diff --git a/tests/test_api_ui_contract.py b/tests/test_api_ui_contract.py index 24f0c73..4397dc4 100644 --- a/tests/test_api_ui_contract.py +++ b/tests/test_api_ui_contract.py @@ -137,3 +137,62 @@ def test_audit_lists_site_entries_newest_first(api): def test_exported_schemas_are_current(): subprocess.run([sys.executable, "scripts/export_schemas.py", "--check"], cwd=ROOT, check=True) + + +def test_run_fed_stats_releases_valid_statistics(api): + client = api[0] + body = _wait(client, client.post("/run", json={"analysis_type": "fed_stats", "variables": ["age", "bmi"], + "project_id": PROJECT}).json()["run_id"]) + RunStatus.model_validate(body) + assert body["status"] == "completed" and body["decision"] == "OK", body + age = body["result"]["stats"]["age"] + assert age["count"] == GT["n_total"] and abs(age["mean"] - GT["mean"]["age"]) < 1e-9 + + +@pytest.mark.parametrize("decision", ["approve", "reject"]) +def test_overseer_decision_updates_the_polled_run(api, decision): + client = api[0] + run_id = client.post("/run", json={"analysis_type": "allele_freq", "variables": ["snp_rs001"], "project_id": PROJECT, + "min_cell_size": 100}).json()["run_id"] + body = _wait(client, run_id) + assert body["status"] == "flagged" + assert client.post(f"/overseer/{body['spec_hash']}/{decision}", json={"by": "tester"}).status_code == 200 + after = client.get(f"/run/{run_id}").json() + RunStatus.model_validate(after) + if decision == "approve": + assert after["status"] == "completed" and after["decision"] == "APPROVED" + ReleasedResult.model_validate(after["result"]) + else: + assert after["status"] == "rejected" and after["decision"] == "REJECTED" and "result" not in after + + +@pytest.mark.parametrize("decision", ["approve", "reject"]) +def test_decision_between_record_and_run_update_is_not_lost(api, monkeypatch, decision): + """record() makes the flagged item visible before the background thread writes the + run's status; a decision landing in that gap must still reach GET /run/{id}.""" + import threading + + from server import api as api_module + + client = api[0] + recorded, proceed = threading.Event(), threading.Event() + real = api_module._analyse + + def analyse_then_pause(*args, **kwargs): + result = real(*args, **kwargs) # record() has run: the item is queued + recorded.set() + proceed.wait(30) + return result + + monkeypatch.setattr(api_module, "_analyse", analyse_then_pause) + submitted = client.post("/run", json={"analysis_type": "allele_freq", "variables": ["snp_rs001"], + "project_id": PROJECT, "min_cell_size": 100}).json() + assert recorded.wait(30) + assert client.post(f"/overseer/{submitted['spec_hash']}/{decision}", json={"by": "tester"}).status_code == 200 + proceed.set() + body = _wait(client, submitted["run_id"]) + if decision == "approve": + assert body["status"] == "completed" and body["decision"] == "APPROVED", body + ReleasedResult.model_validate(body["result"]) + else: + assert body["status"] == "rejected" and body["decision"] == "REJECTED" and "result" not in body, body diff --git a/tests/test_limit_server_jobs.py b/tests/test_limit_server_jobs.py new file mode 100644 index 0000000..a9971fc --- /dev/null +++ b/tests/test_limit_server_jobs.py @@ -0,0 +1,108 @@ +"""scripts/limit_server_jobs.py: synthetic kit layout in tmp_path, no network, no FLARE +provisioning. Run only this file: .venv/bin/python -m pytest -q -p no:cacheprovider +tests/test_limit_server_jobs.py""" +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from scripts.limit_server_jobs import find_server_kit_default, limit_max_jobs + +# Shape matches nvflare's local_server_resources template (master_template.yml): extra +# components and top-level keys included to prove the rest of the document is preserved. +DEFAULT_DOC = { + "format_version": 2, + "class_list_enforcement_mode": "enforce", + "class_allow_list": [], + "servers": [{"admin_storage": "transfer", "max_num_clients": 100, "heart_beat_timeout": 600}], + "snapshot_persistor": { + "path": "nvflare.app_common.state_persistors.storage_state_persistor.StorageStatePersistor", + "args": {"uri_root": "/"}, + }, + "components": [ + { + "id": "job_scheduler", + "path": "nvflare.app_common.job_schedulers.job_scheduler.DefaultJobScheduler", + "args": {"max_jobs": 4}, + }, + { + "id": "job_manager", + "path": "nvflare.apis.impl.job_def_manager.SimpleJobDefManager", + "args": {"uri_root": "/tmp/nvflare/jobs-storage", "job_store_id": "job_store"}, + }, + ], +} + +# A client kit's local/resources.json.default has no job_scheduler component. +CLIENT_DOC = { + "format_version": 2, + "components": [ + { + "id": "resource_manager", + "path": "nvflare.app_common.resource_managers.gpu_resource_manager.GPUResourceManager", + "args": {"num_of_gpus": 0, "mem_per_gpu_in_GiB": 0}, + }, + ], +} + + +def _write_kit(prod_dir: Path, participant: str, doc: dict) -> Path: + local = prod_dir / participant / "local" + local.mkdir(parents=True) + p = local / "resources.json.default" + p.write_text(json.dumps(doc, indent=2)) + return p + + +def test_preserves_full_document_and_only_changes_max_jobs(tmp_path): + prod_dir = tmp_path / "prod_00" + _write_kit(prod_dir, "server_org", DEFAULT_DOC) + _write_kit(prod_dir, "site_a", CLIENT_DOC) + + target = limit_max_jobs(find_server_kit_default(prod_dir)) + + result = json.loads(target.read_text()) + expected = json.loads(json.dumps(DEFAULT_DOC)) # deep copy + expected["components"][0]["args"]["max_jobs"] = 1 + assert result == expected + + +def test_default_file_left_untouched(tmp_path): + prod_dir = tmp_path / "prod_00" + default_path = _write_kit(prod_dir, "server_org", DEFAULT_DOC) + before = default_path.read_text() + + limit_max_jobs(default_path) + + assert default_path.read_text() == before + assert json.loads(before)["components"][0]["args"]["max_jobs"] == 4 + + +def test_idempotent_rerun(tmp_path): + prod_dir = tmp_path / "prod_00" + default_path = _write_kit(prod_dir, "server_org", DEFAULT_DOC) + + first = limit_max_jobs(default_path).read_text() + second = limit_max_jobs(default_path).read_text() + + assert first == second + + +def test_fails_loudly_when_no_job_scheduler_found(tmp_path): + prod_dir = tmp_path / "prod_00" + _write_kit(prod_dir, "site_a", CLIENT_DOC) + _write_kit(prod_dir, "site_b", CLIENT_DOC) + + with pytest.raises(SystemExit, match="found 0"): + find_server_kit_default(prod_dir) + + +def test_fails_loudly_when_multiple_job_scheduler_kits_found(tmp_path): + prod_dir = tmp_path / "prod_00" + _write_kit(prod_dir, "server_org", DEFAULT_DOC) + _write_kit(prod_dir, "server_org_2", DEFAULT_DOC) + + with pytest.raises(SystemExit, match="found 2"): + find_server_kit_default(prod_dir) diff --git a/tests/test_m2_adapters.py b/tests/test_m2_adapters.py index fa52155..759224e 100644 --- a/tests/test_m2_adapters.py +++ b/tests/test_m2_adapters.py @@ -153,3 +153,49 @@ def test_result_roundtrips_as_dict(adapters): spec = AnalysisSpec(analysis_type="allele_freq", variables=["snp_rs001"], project_id=PROJECT) r = next(iter(adapters.values())).run(spec) assert AggregateResult.from_dict(json.loads(json.dumps(r.to_dict()))) == r + + +def test_policy_floor_overrides_requested_min_cell_size(adapters): + spec = AnalysisSpec(analysis_type="allele_freq", variables=["snp_x"], min_cell_size=1, project_id=PROJECT) + r = safe_output.filter("x", "r", spec, 53, {"snp_x": {"genotype_counts": {"0": 50, "1": 3, "2": 0}}}) + assert r.stats["snp_x"].genotype_counts == {"0": 50} + assert "snp_x.genotype_counts.1:count=3<5" in r.rejected + assert _audit_lines("x")[-1]["min_cell_size"] == 5 + + +# ---- missing values: regression inputs must be complete ------------------------------- +def _client_with_missing_values(site, tmp_path, canonical): + """This site's real data with a few values of one variable blanked out.""" + import importlib + + import pandas as pd + from fastapi.testclient import TestClient + + from harmonisation import local_name + + df = pd.read_csv(ROOT / "data" / "sites" / f"{site['tre_id']}.csv") + df.loc[df.index[:3], local_name(canonical, site["tre_id"])] = float("nan") + path = tmp_path / f"{site['tre_id']}.csv" + df.to_csv(path, index=False) + return TestClient(importlib.import_module(f"tres.{site['adapter']}.app").create_app(site["tre_id"], str(path))) + + +def test_incomplete_regression_inputs_are_rejected_on_every_backend(adapters, sites, tmp_path): + feats, outcome = GT["ols"]["features"], GT["ols"]["outcome"] + spec = AnalysisSpec(analysis_type="fed_linreg", variables=feats, outcome=outcome, project_id=PROJECT) + for s in sites: + a = registry.load(s["tre_id"], client=_client_with_missing_values(s, tmp_path, "age")) + r = a.run(spec) + assert (r.n, r.stats, r.rejected) == (0, {}, ["_linreg.gram:incomplete_inputs"]), s["tre_id"] + assert _audit_lines(s["tre_id"])[-1]["decision"] == "REJECTED" + step = a.irls_step(a.local("case"), [a.local(c) for c in feats], [0.0] * (len(feats) + 1), {}) + assert step["complete"] is False and step.get("grad") is None, s["tre_id"] + + +def test_a_variable_with_fewer_than_k_observed_values_releases_nothing(adapters): + spec = AnalysisSpec(analysis_type="fed_stats", variables=["age", "snp_x"], project_id=PROJECT) + raw = {"age": {"count": 1, "sum": 70.0, "sum_sq": 4900.0, "min": 70.0, "max": 70.0}, # one observed age + "snp_x": {"genotype_counts": {}}} # no observed genotype at all + r = safe_output.filter("x", "r", spec, 10, raw) + assert r.stats["age"].present() == {} and r.stats["snp_x"].present() == {} + assert r.rejected == ["age.*:observed=1<5", "snp_x.*:observed=0<5"] diff --git a/tests/test_m3_server.py b/tests/test_m3_server.py index 6c6684a..f1c2907 100644 --- a/tests/test_m3_server.py +++ b/tests/test_m3_server.py @@ -1,5 +1,6 @@ """M3: server merge, disclosure check, overseer queue -- pure unit tests (no FLARE).""" import json +import threading import pytest @@ -71,3 +72,130 @@ def test_overseer_queue_roundtrip(tmp_path, monkeypatch): assert [e["decision"] for e in log] == ["QUEUED", "RELEASED"] and log[-1]["by"] == "tester" with pytest.raises(KeyError): overseer_queue.decide("abc", approve=False, note="", by="x") + + +# ---- policy floor and dominance beyond count tables ------------------------------------ +def test_requested_min_cell_size_cannot_undercut_the_policy_floor(): + m = combine([_r("a", 100, 80, 18, 2), _r("b", 100, 70, 29, 1)], ["a", "b"]) # merged cell '2' = 3 + assert "k_anon:snp_x.genotype_counts.2=3<5" in disclosure_check.check(m, {**SPEC, "min_cell_size": 1})["reasons"] + + +def _stats_site(tid, n, count): + return AggregateResult(tre_id=tid, n=n, region="r", spec_hash="abc", + stats={"age": {"count": count, "sum": 50.0 * count, "sum_sq": 2600.0 * count, + "min": 20.0, "max": 80.0}}) + + +def test_dominance_uses_each_variables_observed_count(): + # equal cohorts (50/50), but one site supplies 100 of the 105 observed ages + m = combine([_stats_site("a", 100, 100), _stats_site("b", 100, 5)], ["a", "b"]) + r = disclosure_check.check(m, {**SPEC, "analysis_type": "fed_stats", "variables": ["age"]}) + assert "dominance:age.count:a=0.95>0.9" in r["reasons"] + + +def _gram_site(tid, n): + matrix = [[1.0 * n, 2.0 * n, 1.0 * n], [2.0 * n, 5.0 * n, 2.5 * n], [1.0 * n, 2.5 * n, 2.0 * n]] + return AggregateResult(tre_id=tid, n=n, region="r", spec_hash="abc", + stats={"_linreg": {"gram": {"n": n, "cols": ["intercept", "y", "x"], "matrix": matrix}}}) + + +def test_dominance_on_regression_sample_counts(): + spec = {**SPEC, "analysis_type": "fed_linreg", "variables": ["x"], "outcome": "y"} + m = combine([_gram_site("a", 950), _gram_site("b", 50)], ["a", "b"]) + assert "dominance:_linreg.n:a=0.95>0.9" in disclosure_check.check(m, spec)["reasons"] + m = combine([_gram_site("a", 500), _gram_site("b", 500)], ["a", "b"]) + assert not any(x.startswith("dominance") for x in disclosure_check.check(m, spec)["reasons"]) + + +def test_dominance_checks_every_released_round(): + # 90/5/5 is not dominated, but a round the last site missed is: 90/95 + merged = {"sites_reported": ["a", "b", "c"], "stats": {}, "n": 100, + "contributions": {"_rounds": {"1": {"a": 90, "b": 5, "c": 5}, "2": {"a": 90, "b": 5}}}} + r = disclosure_check.check(merged, {**SPEC, "analysis_type": "fed_logreg", "variables": ["x"], "outcome": "y"}) + assert r["reasons"] == ["dominance:round=2:a=0.95>0.9"] + + +# ---- release lifecycle: several runs of one spec hash ---------------------------------- +OK = {"decision": "OK", "reasons": [], "signature": "allele_freq|snp_x"} +FLAGGED = {"decision": "FLAGGED", "reasons": ["k_anon:x"], "signature": "allele_freq|snp_x"} + + +def _merged(spec_hash="abc"): + return {**combine([_r("a", 100, 80, 18, 2), _r("b", 100, 70, 25, 5)], ["a", "b"]), "spec_hash": spec_hash} + + +def test_flagged_rerun_withdraws_the_previous_release(tmp_path, monkeypatch): + monkeypatch.setenv("SERVER_OUT", str(tmp_path)) + d = overseer_queue.record(SPEC, _merged(), OK) + assert (d / "released.json").exists() + overseer_queue.record(SPEC, _merged(), FLAGGED) + assert not (d / "released.json").exists() + assert [i["spec_hash"] for i in overseer_queue._load_queue()] == ["abc"] + + +def test_released_rerun_clears_the_stale_queue_entry(tmp_path, monkeypatch): + monkeypatch.setenv("SERVER_OUT", str(tmp_path)) + overseer_queue.record(SPEC, _merged(), FLAGGED) + d = overseer_queue.record(SPEC, _merged(), OK) + assert overseer_queue._load_queue() == [] and (d / "released.json").exists() + + +def test_rejection_leaves_nothing_released(tmp_path, monkeypatch): + monkeypatch.setenv("SERVER_OUT", str(tmp_path)) + overseer_queue.record(SPEC, _merged(), OK) + d = overseer_queue.record(SPEC, _merged(), FLAGGED) + overseer_queue.decide("abc", approve=False, note="", by="tester") + assert not (d / "released.json").exists() and overseer_queue._load_queue() == [] + + +def test_concurrent_decisions_do_not_lose_queue_updates(tmp_path, monkeypatch): + monkeypatch.setenv("SERVER_OUT", str(tmp_path)) + hashes = [f"h{i:02d}" for i in range(20)] + for h in hashes: + overseer_queue.record(SPEC, _merged(h), FLAGGED) + errors = [] + + def decide(h): + try: + overseer_queue.decide(h, True, "", "tester") + except Exception as e: # collected: a torn queue read surfaces here as JSONDecodeError + errors.append(e) + + threads = [threading.Thread(target=decide, args=(h,)) for h in hashes] + for t in threads: + t.start() + for t in threads: + t.join() + assert errors == [] and overseer_queue._load_queue() == [] + assert not list(tmp_path.glob("*.tmp")) # atomic replacement leaves no temp files behind + + +def test_every_released_round_needs_the_minimum_number_of_sites(): + # two sites joined, but round 2's coefficients come from one site alone + merged = {"sites_reported": ["a", "b"], "stats": {}, "n": 100, + "contributions": {"_rounds": {"1": {"a": 50, "b": 50}, "2": {"a": 50}}}} + r = disclosure_check.check(merged, {**SPEC, "analysis_type": "fed_logreg", "variables": ["x"], "outcome": "y"}) + assert r["reasons"] == ["min_sites:round=2:1<2"] + + +def test_released_suppressions_name_the_location_but_not_the_count(tmp_path, monkeypatch): + monkeypatch.setenv("SERVER_OUT", str(tmp_path)) + m = combine([_r("a", 100, 80, 18, 2, rejected=["snp_x.genotype_counts.2:count=2<5"]), _r("b", 100, 70, 25, 5)], ["a", "b"]) + overseer_queue.record(SPEC, m, disclosure_check.check(m, SPEC)) # flagged: site_suppression + d = overseer_queue.decide("abc", approve=True, note="", by="tester") + assert json.loads((d / "released.json").read_text())["rejected_per_site"] == {"a": ["snp_x.genotype_counts.2"]} + # the internal record keeps the detail for the overseer and the audit trail + assert json.loads((d / "result.json").read_text())["rejected_per_site"] == {"a": ["snp_x.genotype_counts.2:count=2<5"]} + + +def test_atomic_writes_keep_normal_file_permissions(tmp_path, monkeypatch): + import os + import stat + + monkeypatch.setenv("SERVER_OUT", str(tmp_path)) + d = overseer_queue.record(SPEC, _merged(), OK) + overseer_queue.record({**SPEC, "min_cell_size": 6}, _merged("other"), FLAGGED) + umask = os.umask(0) + os.umask(umask) + for path in (d / "released.json", overseer_queue.queue_path()): + assert stat.S_IMODE(path.stat().st_mode) == 0o666 & ~umask, path diff --git a/tests/test_m4_linreg.py b/tests/test_m4_linreg.py index c22fb37..295ff29 100644 --- a/tests/test_m4_linreg.py +++ b/tests/test_m4_linreg.py @@ -57,3 +57,34 @@ def test_unstandardise_roundtrip(grams): n, A, b = zip(*[linreg.standardised_normal_eq(g, sc["mean"], sc["std"]) for g in grams]) beta_std = np.linalg.solve(sum(A), sum(b)) assert np.allclose(linreg.unstandardise(beta_std, sc["mean"], sc["std"]), linreg.exact(grams), atol=1e-8) + + +def test_fedavg_controller_does_not_record_a_zero_round_fit(monkeypatch, tmp_path, sites, grams): + """If no training round has two sites, the untouched zero start vector is not a fit.""" + from nvflare.apis.fl_context import FLContext + from nvflare.apis.shareable import Shareable + from nvflare.apis.signal import Signal + + from flare.app.linreg_controller import FedLinregController + from server import overseer_queue + + monkeypatch.setenv("SERVER_OUT", str(tmp_path)) + tids = [s["tre_id"] for s in sites] + + def reply(**fields): + shareable = Shareable() + shareable.update(fields) + return shareable + + def fake_round(name, data, fl_ctx, abort_signal, targets=None): + if name == "linreg_init": + return {tid: reply(moments=linreg.moments(g)) for tid, g in zip(tids, grams)} + upd = linreg.local_update(grams[0], data["beta"], data["mean"], data["std"], data["lr"], data["local_steps"]) + return {tids[0]: reply(update=upd)} # only one site ever trains + + recorded = [] + monkeypatch.setattr(overseer_queue, "record", lambda *args: recorded.append(args)) + ctl = FedLinregController(SPEC.model_dump(), rounds=3, min_clients=1) + monkeypatch.setattr(ctl, "_round", fake_round) + ctl.control_flow(Signal(), FLContext()) + assert recorded == [] diff --git a/tests/test_m6_logreg.py b/tests/test_m6_logreg.py index 800e13b..81c5a48 100644 --- a/tests/test_m6_logreg.py +++ b/tests/test_m6_logreg.py @@ -163,6 +163,9 @@ def fake_round(name, data, fl_ctx, abort_signal, targets=None): res = json.loads((overseer_queue.out_dir() / SPEC.spec_hash() / "result.json").read_text()) MergedResult.model_validate(res) assert res["sites_missing_rounds"] == {"1": [tids[-1]]} + rounds = res["contributions"]["_rounds"] # the dominance check sees who actually answered each round + assert set(rounds["1"]) == set(tids[:-1]) and set(rounds["2"]) == set(tids) + assert rounds["2"] == {tid: GT["n_per_site"][tid] for tid in tids} assert res["method"]["mode"] == "newton_raphson" and res["method"]["rounds"] > 1 assert np.max(np.abs(np.array([res["stats"]["_logreg"]["logreg"]["coef"][k] for k in ("intercept", *FEATURES)]) - _truth())) < 1e-6 @@ -186,3 +189,121 @@ def test_result_contract_accepts_logreg_and_still_rejects_unknown_fields(): MergedResult.model_validate({**merged, "sites_missing_rounds": {1: ["c"]}}) with pytest.raises(ValidationError): Method(mode="newton_raphson", bogus=1) + + +# ---- Safe Output for one Newton round (shared by the FLARE executor and the direct API) ---- +def _audit_tail(): + import os + from pathlib import Path + + return json.loads((Path(os.environ["AUDIT_DIR"]) / "audit.jsonl").read_text().splitlines()[-1]) + + +def _valid_step(n, p): + return {"n": n, "complete": True, "grad": [0.5] * p, "hess": [[float(i == j) for j in range(p)] for i in range(p)]} + + +def _needed(p): + from adapters import safe_output + + # same rule as a Gram matrix over [intercept, outcome, *features] + return max(safe_output.effective_min_cell_size(SPEC.project_id, SPEC.min_cell_size), p + 2) + + +def test_irls_release_accepts_and_audits_a_valid_round(adapters): + from adapters import safe_output + + p = len(FEATURES) + 1 + step = safe_output.release_irls_step("x", SPEC, _valid_step(_needed(p), p), p) + assert step == {k: v for k, v in _valid_step(_needed(p), p).items() if k != "complete"} + assert _audit_tail()["decision"] == "OK:newton_raphson_round" + + +@pytest.mark.parametrize("change,reason", [ + ({"complete": False}, "incomplete_inputs"), + ({"complete": None}, "incomplete_inputs"), # a backend that does not say is not trusted + ({"n": "small"}, "invalid_n"), + ({"grad": [0.5]}, "invalid_shape"), + ({"hess": [[1.0]]}, "invalid_shape"), + ({"grad": [float("nan")] * (len(FEATURES) + 1)}, "non_finite"), +]) +def test_irls_release_rejects_and_audits(adapters, change, reason): + from adapters import safe_output + + p = len(FEATURES) + 1 + with pytest.raises(safe_output.Rejected): + safe_output.release_irls_step("x", SPEC, {**_valid_step(_needed(p), p), **change}, p) + assert _audit_tail()["decision"] == "REJECTED" and _audit_tail()["rejected"] == [f"_logreg.step:{reason}"] + + +def test_irls_release_enforces_the_sample_threshold(adapters): + from adapters import safe_output + + p = len(FEATURES) + 1 + with pytest.raises(safe_output.Rejected): + safe_output.release_irls_step("x", SPEC, _valid_step(_needed(p) - 1, p), p) + assert _audit_tail()["rejected"] == [f"_logreg.step:n={_needed(p) - 1}<{_needed(p)}"] + + +def test_controller_fit_counts_only_sites_that_contributed_a_round(monkeypatch, tmp_path, sites, adapters): + from nvflare.apis.fl_context import FLContext + from nvflare.apis.shareable import Shareable + from nvflare.apis.signal import Signal + + from flare.app.logreg_controller import FedLogregController + from server import overseer_queue + + monkeypatch.setenv("SERVER_OUT", str(tmp_path)) + tids = [s["tre_id"] for s in sites] + + def reply(**fields): + shareable = Shareable() + shareable.update(fields) + return shareable + + def fake_round(name, data, fl_ctx, abort_signal, targets=None): + if name == "logreg_init": + return {tid: reply(n=GT["n_per_site"][tid]) for tid in tids} + return {tid: reply(step=_step(adapters[tid], data["beta"])) for tid in tids[:-1]} # the last never answers + + ctl = FedLogregController(SPEC.model_dump(), rounds=25, min_clients=2) + monkeypatch.setattr(ctl, "_round", fake_round) + ctl.control_flow(Signal(), FLContext()) + res = json.loads((overseer_queue.out_dir() / SPEC.spec_hash() / "result.json").read_text()) + assert res["sites_reported"] == sorted(tids[:-1]) and tids[-1] in res["sites_missing"] + assert res["n"] == res["stats"]["_logreg"]["n"] == sum(GT["n_per_site"][t] for t in tids[:-1]) + assert res["n_per_site"] == {t: GT["n_per_site"][t] for t in tids[:-1]} + + +def test_controller_never_uses_a_single_site_round(monkeypatch, tmp_path, sites, adapters): + """Even with min_clients=1 (a supported job setting), a round answered by one site + would be that site's own Newton step: the fit stops instead of releasing it.""" + from nvflare.apis.fl_context import FLContext + from nvflare.apis.shareable import Shareable + from nvflare.apis.signal import Signal + + from flare.app.logreg_controller import FedLogregController + from server import overseer_queue + + monkeypatch.setenv("SERVER_OUT", str(tmp_path)) + tids = [s["tre_id"] for s in sites] + calls = {"n": 0} + + def reply(**fields): + shareable = Shareable() + shareable.update(fields) + return shareable + + def fake_round(name, data, fl_ctx, abort_signal, targets=None): + if name == "logreg_init": + return {tid: reply(n=GT["n_per_site"][tid]) for tid in tids} + calls["n"] += 1 + answering = tids if calls["n"] == 1 else tids[:1] # only one site answers from round 2 on + return {tid: reply(step=_step(adapters[tid], data["beta"])) for tid in answering} + + ctl = FedLogregController(SPEC.model_dump(), rounds=25, min_clients=1) + monkeypatch.setattr(ctl, "_round", fake_round) + ctl.control_flow(Signal(), FLContext()) + res = json.loads((overseer_queue.out_dir() / SPEC.spec_hash() / "result.json").read_text()) + assert [h["sites"] for h in res["stats"]["_logreg"]["history"]] == [len(tids)] + assert list(res["contributions"]["_rounds"]) == ["1"] diff --git a/tres/common.py b/tres/common.py index 4e15197..ace8c7e 100644 --- a/tres/common.py +++ b/tres/common.py @@ -75,12 +75,20 @@ def aggregate(series: pd.Series, agg: str) -> Any: raise ValueError(f"unsupported agg {agg!r}") +def _complete(df: pd.DataFrame, cols: list[str]) -> bool: + """Regression inputs must be complete: a row missing any column would otherwise + poison the sums (NaN) or, in SQL, silently drop out of some sums but not others.""" + return bool(df[cols].notna().all(axis=None)) + + def gram(df: pd.DataFrame, cols: list[str]) -> dict: """Cross-product matrix of [1, *cols]; the sufficient statistic for OLS / means / variances.""" import numpy as np + if not _complete(df, cols): + return {"n": int(len(df)), "cols": ["1", *cols], "matrix": None, "complete": False} X = np.column_stack([np.ones(len(df))] + [df[c].to_numpy(float) for c in cols]) - return {"n": int(len(df)), "cols": ["1", *cols], "matrix": (X.T @ X).tolist()} + return {"n": int(len(df)), "cols": ["1", *cols], "matrix": (X.T @ X).tolist(), "complete": True} def irls_step(df: pd.DataFrame, outcome: str, features: list[str], beta: list[float]) -> dict: @@ -90,6 +98,8 @@ def irls_step(df: pd.DataFrame, outcome: str, features: list[str], beta: list[fl the TRE process, to produce this aggregate; only n/grad/hess ever leave.""" import numpy as np + if not _complete(df, [outcome, *features]): + return {"n": int(len(df)), "complete": False, "grad": None, "hess": None} X = np.column_stack([np.ones(len(df))] + [df[c].to_numpy(float) for c in features]) y = df[outcome].to_numpy(float) eta = X @ np.asarray(beta, dtype=float) @@ -97,4 +107,4 @@ def irls_step(df: pd.DataFrame, outcome: str, features: list[str], beta: list[fl w = np.clip(p * (1 - p), 1e-6, None) grad = X.T @ (y - p) hess = X.T @ (X * w[:, None]) - return {"n": int(len(df)), "grad": grad.tolist(), "hess": hess.tolist()} + return {"n": int(len(df)), "complete": True, "grad": grad.tolist(), "hess": hess.tolist()}