diff --git a/.env.development.example b/.env.development.example index 6bcbdac57..3e349c995 100644 --- a/.env.development.example +++ b/.env.development.example @@ -341,6 +341,11 @@ FL_APP_DESTINATION_BUCKET=s3://${FLIP_APP_BUNDLES_BUCKET_NAME}/app_destinations # BANDIT_TIMEOUT_SECONDS=60 # How often the sweep re-checks uploads left SCANNING by an app restart. # SCHEDULER_MALWARE_SCAN_RECONCILE_RATE=1 +# How often the hub asks each net's FL API whether an in-flight job has failed +# (FLIP#1001), and how long a job may go unlisted by its backend (a SuperLink +# restart loses its in-memory run state) before it is treated as dead. +# SCHEDULER_FL_JOB_RECONCILE_RATE=1 +# FL_JOB_UNLISTED_GRACE_MINUTES=30 # ── FL Net Endpoints ────────────────────────────────────────────────────── # Maps FL network IDs to internal API URLs for the FL API containers on the diff --git a/AGENTS.md b/AGENTS.md index f8466235d..72315cf63 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -395,6 +395,16 @@ After changes, evaluate if docs need updating: - `PICKLESCAN_FILE_SUFFIXES` / `PICKLESCAN_TIMEOUT_SECONDS` — which uploads get a structural picklescan before promotion (default `.pt .pth .pkl .pickle`) and the wall-clock cap per scan (default 120s). Dangerous globals mark the file `INFECTED` and delete the object; a scan that errors or times out fails closed to `ERROR`. Signature-based AV (GuardDuty Malware Protection for S3) is tracked separately in FLIP#838. - `BANDIT_TIMEOUT_SECONDS` — wall-clock cap (default 60s) on the non-blocking Bandit pass over `.py` uploads (FLIP#877, GHSA-8465). Unlike `PICKLESCAN_TIMEOUT_SECONDS` a timeout here just means no findings are recorded (fail-open, never `ERROR`) — Bandit is advisory only and never gates promotion. Findings land on `UploadedFiles.bandit_findings` (`[]` = scanned clean, `NULL` = never scanned) and surface in the UI as an amber indicator. `bandit` is a dependency baked into the `flip-api` image, not the mounted `src/`: under the dev pull-by-default sourcing above, running without `BUILD=true` after picking up this dependency serves an image with no `bandit` binary, so every upload silently records `NULL` findings while the UI and docs still claim a scan ran. - `SCHEDULER_MALWARE_SCAN_RECONCILE_RATE` — how often (minutes, default 1) the sweep re-checks uploads left `SCANNING` by an app restart mid-scan. +- `SCHEDULER_FL_JOB_RECONCILE_RATE` / `FL_JOB_UNLISTED_GRACE_MINUTES` / `FLOWER_RUN_LOG_MAX_CHARS` — how often + (minutes, default 1) the hub asks each net's FL API whether an in-flight job has failed, how long (default 30 min) + a job may go unlisted by its backend before it is treated as dead (SuperLink run state is in-memory, so a restart + forgets every run), and the cap (default 8000 chars) on the run-log tail `fl-api-flower`'s `GET /run_logs/{run_id}` + returns. Nothing reports a run that dies *after* submission (an ImportError at ServerApp module scope, say) — the + model would otherwise sit at `INITIATED` forever — so the sweep (`flip_api/fl_services/reconcile_failed_jobs.py`) + polls for it, moves the model to `ERROR` and stores the log tail on the activity feed. Only `FAILED` and + unlisted-past-grace act; `FINISHED` is left to the run's own `RESULTS_UPLOADED` callback, and `UNKNOWN` (an + unmapped native status — the shared #490 contract's sixth value, never guessed into FAILED) is a no-op. Covers + the ServerApp only — a ClientApp dying at a trust logs to that trust's SuperNode (FLIP#1001). ## Deployment Architecture diff --git a/CLAUDE.md b/CLAUDE.md index 750596df9..f72ad94bc 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -395,6 +395,16 @@ After changes, evaluate if docs need updating: - `PICKLESCAN_FILE_SUFFIXES` / `PICKLESCAN_TIMEOUT_SECONDS` — which uploads get a structural picklescan before promotion (default `.pt .pth .pkl .pickle`) and the wall-clock cap per scan (default 120s). Dangerous globals mark the file `INFECTED` and delete the object; a scan that errors or times out fails closed to `ERROR`. Signature-based AV (GuardDuty Malware Protection for S3) is tracked separately in FLIP#838. - `BANDIT_TIMEOUT_SECONDS` — wall-clock cap (default 60s) on the non-blocking Bandit pass over `.py` uploads (FLIP#877, GHSA-8465). Unlike `PICKLESCAN_TIMEOUT_SECONDS` a timeout here just means no findings are recorded (fail-open, never `ERROR`) — Bandit is advisory only and never gates promotion. Findings land on `UploadedFiles.bandit_findings` (`[]` = scanned clean, `NULL` = never scanned) and surface in the UI as an amber indicator. `bandit` is a dependency baked into the `flip-api` image, not the mounted `src/`: under the dev pull-by-default sourcing above, running without `BUILD=true` after picking up this dependency serves an image with no `bandit` binary, so every upload silently records `NULL` findings while the UI and docs still claim a scan ran. - `SCHEDULER_MALWARE_SCAN_RECONCILE_RATE` — how often (minutes, default 1) the sweep re-checks uploads left `SCANNING` by an app restart mid-scan. +- `SCHEDULER_FL_JOB_RECONCILE_RATE` / `FL_JOB_UNLISTED_GRACE_MINUTES` / `FLOWER_RUN_LOG_MAX_CHARS` — how often + (minutes, default 1) the hub asks each net's FL API whether an in-flight job has failed, how long (default 30 min) + a job may go unlisted by its backend before it is treated as dead (SuperLink run state is in-memory, so a restart + forgets every run), and the cap (default 8000 chars) on the run-log tail `fl-api-flower`'s `GET /run_logs/{run_id}` + returns. Nothing reports a run that dies *after* submission (an ImportError at ServerApp module scope, say) — the + model would otherwise sit at `INITIATED` forever — so the sweep (`flip_api/fl_services/reconcile_failed_jobs.py`) + polls for it, moves the model to `ERROR` and stores the log tail on the activity feed. Only `FAILED` and + unlisted-past-grace act; `FINISHED` is left to the run's own `RESULTS_UPLOADED` callback, and `UNKNOWN` (an + unmapped native status — the shared #490 contract's sixth value, never guessed into FAILED) is a no-op. Covers + the ServerApp only — a ClientApp dying at a trust logs to that trust's SuperNode (FLIP#1001). ## Deployment Architecture diff --git a/docs/source/components/component-fl-nodes.rst b/docs/source/components/component-fl-nodes.rst index 0f4ae60a6..7cd402a89 100644 --- a/docs/source/components/component-fl-nodes.rst +++ b/docs/source/components/component-fl-nodes.rst @@ -293,6 +293,53 @@ Note the reported upload sizes measure slightly different things per backend — The server will also use the package to update the status, as well as to upload the final results, which will be first saved in the server, to the final S3 buckets users can download from. +When a run fails after submission +--------------------------------- + +Almost everything the Central Hub knows about a run in progress is reported *by* the run, +through the ``flip`` package above. A run that dies before it can report anything — the +classic case is an exception at import time in ``server_app.py``, which kills the ServerApp +the instant it starts — would therefore say nothing at all, leaving the model showing +``INITIATED`` indefinitely. + +The hub closes that gap by polling. Once a minute it asks each net's FL API for the status of +every job it still believes is in flight; a job the backend reports as failed moves the model +to ``ERROR``, frees the net, and writes the tail of the run's own log to the model's activity +feed, where the researcher reads it alongside the round events. A job the backend no longer +lists at all is treated the same way once it has gone unlisted for a grace period +(``FL_JOB_UNLISTED_GRACE_MINUTES``, default 30) — the Flower SuperLink keeps run state in +memory, so a restart forgets every run, and such a run can never report. Nothing else is +acted on: a run that is pending, running or finished is left entirely to its own reporting. + +The feed row leads with the backend's own one-line explanation of the failure, when it has +one. Flower reports this on every ``flwr ls`` entry, so the hub gets the ServerApp's exception +type and message from the same call it already makes to read the status — no extra request. +NVFLARE has no equivalent field, so its rows carry the status alone. + +Two things are worth knowing about the captured log. It is a **tail**: a Flower run log opens +with the per-run dependency install and the cause of a failure is at the far end, so the head +is dropped. And it is best-effort — if the log cannot be retrieved, the feed says so and +names the manual fallback. On a Flower net that is to run, inside the net's FL API container: + +.. code-block:: bash + + flwr log local --show + +On an NVFLARE net (whose FL API serves no run-log endpoint today, so its failures always +surface status-only) the fallback is the fl-server container's own output: + +.. code-block:: bash + + docker logs fl-server-net- + +NVFLARE keeps no readable per-job log in the job workspace — the workspace holds opaque +archive blobs — so the container output is where the traceback actually lives. + +This covers the **ServerApp**. A ClientApp that dies at a trust writes to that trust's +SuperNode, which the Central Hub cannot read; such failures still surface only through +whatever the app itself reports before dying. + + Privacy filters on shared model updates --------------------------------------- diff --git a/fl-services/flower/README.md b/fl-services/flower/README.md index 2c09c66c6..92201f7bf 100644 --- a/fl-services/flower/README.md +++ b/fl-services/flower/README.md @@ -150,3 +150,25 @@ make -C fl-services/flower down The central-hub multi-net Flower topology is the separate [`deploy/compose.development.flower.yml`](../../deploy/compose.development.flower.yml), driven by the root `make up FL_BACKEND=flower`. + +## Diagnosing a run that fails after submission + +A Flower run whose ServerApp dies logs almost nothing where you would look for it: the SuperLink +reports `Started task` and `Finished task` for the run in the same breath and says no more, and no +container's `docker logs` carries the traceback. The run's own log lives in the SuperLink, reachable +only through the Control API. + +The Central Hub polls for this (FLIP#1001): a job the FL API reports as failed drives the model to +`ERROR` and the tail of the run log is written to the model's activity feed, so the usual first stop +is the model page rather than a shell. To read the full stream by hand — or to inspect a run +submitted outside the hub, e.g. via `make submit` — exec into the net's FL API container: + +```bash +docker exec -it flip-fl-api-net-1 uvx flwr log local --show # hub multi-net stack +docker compose -f fl-services/flower/compose.dev.yml exec fl-api \ + uvx flwr log local --show # standalone dev stack +``` + +`--show` prints the stored log and exits; the `flwr log` default (`--stream`) follows it forever. +Run ids come from `uvx flwr list local` in the same container, or from the `run-id` the submit +returned. A ServerApp that died at import time ends with a traceback and `ERROR: Exit Code: 607`. diff --git a/fl-services/flower/fl-api-flower/README.md b/fl-services/flower/fl-api-flower/README.md index 6b49d9e91..aa4f0ee70 100644 --- a/fl-services/flower/fl-api-flower/README.md +++ b/fl-services/flower/fl-api-flower/README.md @@ -27,6 +27,7 @@ Standalone FastAPI service for Flower deployment runtime. - `POST /submit_run/{job_folder}` — submit a previously uploaded application; `job_folder` is the Central Hub `model_id` (UUID). flip-api's production path (also exposed as the hidden `/submit_job` alias) - `POST /submit_tutorial/{tutorial_name}` — submit a pre-baked tutorial folder by name (e.g. `numpy`, `xray_classification`); the local tutorial harness targets this - `DELETE /abort_run/{run_id}` +- `GET /run_logs/{run_id}` — a bounded, secret-masked tail of a run's ServerApp log; the Central Hub reads it when it finds a run in a failed state (FLIP#1001) ## API docs @@ -63,6 +64,27 @@ uvx flwr stop local --format json It returns the full JSON payload from Flower. +The run-logs endpoint runs: + +```bash +uvx flwr log local --show +``` + +`--show` (rather than the `flwr log` default `--stream`, which follows the log forever) prints what +the SuperLink has stored for the run and exits. The response is +`{"run_id": ..., "log": ..., "truncated": ...}`, where `log` is the **last** `FLOWER_RUN_LOG_MAX_CHARS` +characters (default 8000) of the output: a Flower run log opens with the per-run dependency install and +the cause of a failure is at the other end, so the head is the half worth dropping. Credential-shaped +substrings are masked first — a run log is whatever researcher-supplied ServerApp code printed, in a +container that holds a hub service key, so it is not trusted to be secret-free. + +To read the full stream by hand instead, exec into this container and run the same command without +the truncation: + +```bash +docker exec -it flip-fl-api-net-1 uvx flwr log local --show +``` + The server status endpoint checks the Flower SuperLink health service configured by `SUPERLINK_HEALTH_ADDRESS` and returns: @@ -87,6 +109,8 @@ If `targets` are omitted, all registered trust names are returned. Set these environment variables in the FL API container: - `SUPERLINK_HEALTH_ADDRESS` (example: `superlink:9097`) — for server status checks +- `FLOWER_RUN_LOG_MAX_CHARS` (optional, default `8000`) — cap on the run-log tail returned by + `/run_logs/{run_id}`. An unset, empty or unparseable value falls back to the default. ## Development startup with Docker Compose diff --git a/fl-services/flower/fl-api-flower/fl_api/app.py b/fl-services/flower/fl-api-flower/fl_api/app.py index 6488a72d2..03807fe9f 100644 --- a/fl-services/flower/fl-api-flower/fl_api/app.py +++ b/fl-services/flower/fl-api-flower/fl_api/app.py @@ -34,10 +34,12 @@ JobMetadata, JobStatus, NodeRegistrationRequest, + RunLogs, ServerInfoModel, UploadAppRequest, normalize_status, ) +from fl_api.utils.redaction import redact_secrets from fl_api.utils.upload import upload_application from fl_api.utils.validation import safe_join, validate_tutorial_folder_name @@ -59,6 +61,20 @@ _node_mapping_lock = threading.Lock() _node_trust_mapping: dict[str, str] = {} # Flower node_id → trust name +# `flwr log --show` asks the SuperLink for the run's stored log and returns; its own gRPC +# deadline is 5s, so anything past this means the CLI itself is wedged (unreachable +# SuperLink, uvx resolving the package) rather than a slow run. +_RUN_LOG_COMMAND_TIMEOUT_SECONDS = 60 +# Cap on the `status-details` one-liner carried on each list_jobs item. It is a status +# string, not a log, so this is a sanity bound rather than a real truncation policy — but +# the text is a researcher-authored exception message, so it is not trusted to be short. +_MAX_STATUS_DETAILS_CHARS = 500 +# `flwr ls` writes this literal for a run with nothing to say (a healthy or still-running +# one), rather than omitting the key. Carrying it through would put "N/A" in the hub's +# activity feed as though it were a cause. +_NO_STATUS_DETAILS = "N/A" +_DEFAULT_RUN_LOG_MAX_CHARS = 8000 + def _get_src_root() -> Path: return Path(os.getenv("FLOWER_SRC_ROOT", "/app/src")) @@ -83,6 +99,24 @@ def _get_healthcheck_timeout_seconds() -> float: return 1.0 +def _get_run_log_max_chars() -> int: + raw_value = os.getenv("FLOWER_RUN_LOG_MAX_CHARS", "").strip() + if not raw_value: + return _DEFAULT_RUN_LOG_MAX_CHARS + try: + max_chars = int(raw_value) + if max_chars <= 0: + raise ValueError("max chars must be positive") + return max_chars + except ValueError: + logger.warning( + "Invalid FLOWER_RUN_LOG_MAX_CHARS='%s'. Falling back to %d characters.", + raw_value, + _DEFAULT_RUN_LOG_MAX_CHARS, + ) + return _DEFAULT_RUN_LOG_MAX_CHARS + + def _check_health(address: str, timeout: float) -> bool: if not address: logger.warning("Health check address is empty.") @@ -121,7 +155,9 @@ def _extract_json_from_stdout(stdout: str) -> dict[str, Any]: return parsed -def _run_flwr_command(command: list[str], cwd: Path, action_name: str) -> subprocess.CompletedProcess[str]: +def _run_flwr_command( + command: list[str], cwd: Path, action_name: str, timeout: float | None = None +) -> subprocess.CompletedProcess[str]: try: return subprocess.run( command, @@ -129,6 +165,7 @@ def _run_flwr_command(command: list[str], cwd: Path, action_name: str) -> subpro capture_output=True, text=True, check=False, + timeout=timeout, ) except Exception as err: logger.exception("Failed to run Flower %s command", action_name) @@ -194,6 +231,28 @@ def _get_federation_nodes(src_root: Path) -> list[dict[str, Any]]: return nodes +def _clean_status_details(raw: Any) -> str | None: + """Normalise one run's ``status-details`` into the contract's optional one-liner. + + Redacted like the run log is, and for the same reason: the text is whatever the + ServerApp's exception carried, written by researcher-supplied code in a container that + holds a hub service key. Collapsed to a single line so it can be used as a headline in + the hub's activity feed without swallowing the log tail that follows it. + + Args: + raw (Any): The ``status-details`` value from one `flwr ls` entry, if any. + + Returns: + str | None: The cleaned one-liner, or None when the backend had nothing to say. + """ + if not isinstance(raw, str): + return None + collapsed = " ".join(raw.split()) + if not collapsed or collapsed == _NO_STATUS_DETAILS: + return None + return redact_secrets(collapsed)[:_MAX_STATUS_DETAILS_CHARS] + + def _parse_runs_payload(payload: dict[str, Any]) -> list[JobMetadata]: runs = payload.get("runs") if not isinstance(runs, list): @@ -211,6 +270,9 @@ def _parse_runs_payload(payload: dict[str, Any]) -> list[JobMetadata]: JobMetadata( job_id=str(run["run-id"]), status=normalize_status(run["status"]), + # Absent on older flwr versions -- .get, so a missing key is simply no + # detail rather than the 500 a missing run-id/status earns below. + status_details=_clean_status_details(run.get("status-details")), ) ) except KeyError as err: @@ -317,6 +379,68 @@ def list_runs() -> list[JobMetadata]: return _parse_runs_payload(payload) +def _tail(text: str, max_chars: int) -> tuple[str, bool]: + """Return the last ``max_chars`` characters of ``text`` and whether anything was dropped. + + Args: + text (str): The full text. + max_chars (int): Maximum number of characters to keep. + + Returns: + tuple[str, bool]: The tail, and True when the head was dropped. The cut is + advanced to the next line boundary so the tail never opens mid-line. + """ + if len(text) <= max_chars: + return text, False + + tail = text[-max_chars:] + newline = tail.find("\n") + if newline != -1: + tail = tail[newline + 1 :] + return tail, True + + +@app.get("/run_logs/{run_id}", status_code=status.HTTP_200_OK, response_model=RunLogs) +def run_logs(run_id: int) -> RunLogs: + """Return a bounded, secret-masked tail of a run's ServerApp log. + + Exists so a run that dies after submission — an import error at ServerApp module + scope, say — can be diagnosed from the Central Hub instead of by exec-ing into this + container and running ``flwr log`` by hand (FLIP#1001). The hub's FL job reconcile + calls this for a run it has found in a failed state and stores the result on the + model's activity feed. + + Args: + run_id (int): The Flower run id. Typed as ``int`` for the same reason as + ``abort_run``: FastAPI rejects any non-numeric segment with 422 before it + can reach the ``flwr`` argv. + + Returns: + RunLogs: The run id, the log tail, and whether the head was dropped. + + Raises: + HTTPException: 500 when the ``flwr log`` command cannot be run or fails. + """ + run_id_str = str(run_id) + # --show prints what the SuperLink has stored for the run and exits; the default + # --stream would follow the log forever and never return to the caller. + command = ["uvx", "flwr", "log", run_id_str, "local", "--show"] + result = _run_flwr_command(command, _get_src_root(), "log", timeout=_RUN_LOG_COMMAND_TIMEOUT_SECONDS) + + if result.returncode != 0: + stderr = result.stderr.strip() + logger.error("Flower log failed for run %s: %s", run_id_str, stderr) + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail=f"Flower log command failed with code {result.returncode}. stderr: {redact_secrets(stderr)}", + ) + + # Redact before truncating, not after: the tail is cut from the middle of the log, and + # a cut landing inside a `key=` pair would strip the keyword the matcher needs. + log_tail, truncated = _tail(redact_secrets(result.stdout), _get_run_log_max_chars()) + return RunLogs(run_id=run_id_str, log=log_tail, truncated=truncated) + + def _submit_from_job_dir(job_dir: Path, label: str) -> str: global _submission_in_progress diff --git a/fl-services/flower/fl-api-flower/fl_api/schemas.py b/fl-services/flower/fl-api-flower/fl_api/schemas.py index 544509453..8b7a97220 100644 --- a/fl-services/flower/fl-api-flower/fl_api/schemas.py +++ b/fl-services/flower/fl-api-flower/fl_api/schemas.py @@ -70,6 +70,11 @@ class JobStatus(StrEnum): FINISHED = "FINISHED" FAILED = "FAILED" STOPPED = "STOPPED" + # A native status the map below does not recognise — i.e. a framework upgrade added one. + # Never guessed into a real state: FAILED now has destructive consumers on the hub (the + # failed-job reconcile errors the model and frees the net), and RUNNING would make the + # job abortable, so an unmapped status must reach the hub as explicitly unknowable. + UNKNOWN = "UNKNOWN" class JobMetadata(BaseModel): @@ -77,6 +82,29 @@ class JobMetadata(BaseModel): job_id: str status: JobStatus + # One-line native explanation of the status, when the backend supplies one. Flower's + # `flwr ls` carries `status-details`, which for a failed run is the ServerApp's exception + # type and message — the cause, from the call the hub already makes, with no second + # request and no log fetch. Optional because it is not universal: NVFLARE's job meta has + # no equivalent field (checked against `nvflare.apis.job_def.JobMetaKey`, which carries + # only `job_deploy_detail` / `schedule_history`, neither of which explains an + # execution exception), so that adapter leaves this None. + status_details: str | None = None + + +class RunLogs(BaseModel): + """The response of ``GET /run_logs/{run_id}`` — a bounded tail of a run's ServerApp output. + + Only the tail is returned: a Flower run log opens with the per-run dependency + install (``uv sync`` over the app's whole dependency set) and the cause of a + failure is at the other end, so the head is the half worth dropping. + """ + + run_id: str + log: str + # True when the stored log was longer than the cap and the head was dropped, so a + # reader knows to go to `flwr log` for the full stream rather than assuming this is all. + truncated: bool # Flower native status (`flwr list` / `flwr stop`) -> normalized contract status. @@ -101,10 +129,11 @@ def normalize_status(native_status: str) -> JobStatus: Returns: JobStatus: The normalized status. Unknown / unmapped statuses are logged and - treated as ``FAILED`` — never silently surfaced as an abortable ``RUNNING``. + surfaced as ``UNKNOWN`` — never guessed into an abortable ``RUNNING`` or an + actionable ``FAILED`` (the hub's failed-job reconcile acts on FAILED). """ normalized = _FLOWER_STATUS_MAP.get(native_status.strip().lower()) if normalized is None: - logger.warning("Unmapped Flower job status %r; treating as FAILED.", native_status) - return JobStatus.FAILED + logger.warning("Unmapped Flower job status %r; surfacing as UNKNOWN.", native_status) + return JobStatus.UNKNOWN return normalized diff --git a/fl-services/flower/fl-api-flower/fl_api/utils/redaction.py b/fl-services/flower/fl-api-flower/fl_api/utils/redaction.py new file mode 100644 index 000000000..459acd0ac --- /dev/null +++ b/fl-services/flower/fl-api-flower/fl_api/utils/redaction.py @@ -0,0 +1,71 @@ +# Copyright (c) 2026 Guy's and St Thomas' NHS Foundation Trust & King's College London +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# http://www.apache.org/licenses/LICENSE-2.0 +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +"""Best-effort secret masking for text that leaves the FL API. + +Applied to Flower run logs before they are handed to the Central Hub (and from +there into ``fl_logs``, which the model owner reads in the UI). A run log is +whatever the ServerApp and its runtime wrote to stdout/stderr — researcher- +supplied code running in a container that holds ``INTERNAL_SERVICE_KEY`` and +AWS credentials — so it is not trusted to be secret-free. This is damage +limitation on a channel that should not carry secrets in the first place, not a +guarantee that none get through: a secret printed without a recognisable +keyword still passes. + +Deliberately biased towards over-redaction. Masking a stray ``max_tokens=512`` +costs a reader nothing; leaking a service key costs a rotation. +""" + +import re + +REDACTED = "[REDACTED]" + +_SECRET_PATTERNS: list[tuple[re.Pattern[str], str]] = [ + # SigV4 query parameters on a presigned S3 URL. Each is a standalone + # capability against the bucket for the life of the signature, so they are + # masked individually rather than dropping the whole URL — the object path + # is the useful half of the line. + ( + re.compile(r"(?i)([?&](?:X-Amz-Signature|X-Amz-Credential|X-Amz-Security-Token)=)[^&\s\"']+"), + rf"\1{REDACTED}", + ), + # `key|token|secret|password|credential` followed by `:` or `=`. + # Covers header dumps (`X-Internal-Service-Key: ...`), env dumps + # (`AWS_SECRET_ACCESS_KEY=...`) and kwargs repr in a traceback frame. + # Requiring the separator immediately after the keyword keeps prose and + # `KeyError: 'x'` out of it; stopping the value at `&` keeps it from eating + # the rest of a query string the rule above has already masked. + ( + re.compile( + r"(?i)((?:[\w-]{0,64}(?:key|token|secret|password|passwd|credential))[\"']?\s*[:=]\s*[\"']?)" + r"[^\s\"',}&]+" + ), + rf"\1{REDACTED}", + ), + # A bare AWS access key id carries no keyword to key off, and is the one + # credential shape distinctive enough to match on its own. + (re.compile(r"\b(?:AKIA|ASIA)[0-9A-Z]{16}\b"), REDACTED), +] + + +def redact_secrets(text: str) -> str: + """Mask credential-shaped substrings in ``text``. + + Args: + text (str): Arbitrary log text. + + Returns: + str: The same text with recognised credentials replaced by ``[REDACTED]``. + """ + for pattern, replacement in _SECRET_PATTERNS: + text = pattern.sub(replacement, text) + return text diff --git a/fl-services/flower/fl-api-flower/tests/conftest.py b/fl-services/flower/fl-api-flower/tests/conftest.py index 6f1259232..ea7f64074 100644 --- a/fl-services/flower/fl-api-flower/tests/conftest.py +++ b/fl-services/flower/fl-api-flower/tests/conftest.py @@ -49,19 +49,25 @@ def reset_node_mapping(): @pytest.fixture def mock_flwr_run(monkeypatch): + # Returns the list of argv lists the app handed to subprocess.run, so a test can + # assert on the command it built as well as on the response. def _mock(*, returncode=0, stdout="", stderr="", exception=None, by_command=None): + commands: list[list[str]] = [] + if exception is not None: - def _raise(*_args, **_kwargs): + def _raise(command=None, *_args, **_kwargs): + commands.append(command) raise exception monkeypatch.setattr(app_module.subprocess, "run", _raise) - return + return commands if by_command is not None: # by_command: {flwr_subcommand: {"returncode": int, "stdout": str, "stderr": str}} # e.g. {"stop": {...}, "list": {...}}. command is ["uvx", "flwr", "", ...]. def _dispatch(command, *_args, **_kwargs): + commands.append(command) subcommand = command[2] if len(command) > 2 else "" spec = by_command.get(subcommand, {}) return subprocess.CompletedProcess( @@ -72,17 +78,18 @@ def _dispatch(command, *_args, **_kwargs): ) monkeypatch.setattr(app_module.subprocess, "run", _dispatch) - return + return commands - monkeypatch.setattr( - app_module.subprocess, - "run", - lambda *_args, **_kwargs: subprocess.CompletedProcess( - args=[], + def _fixed(command=None, *_args, **_kwargs): + commands.append(command) + return subprocess.CompletedProcess( + args=command or [], returncode=returncode, stdout=stdout, stderr=stderr, - ), - ) + ) + + monkeypatch.setattr(app_module.subprocess, "run", _fixed) + return commands return _mock diff --git a/fl-services/flower/fl-api-flower/tests/test_abort_run.py b/fl-services/flower/fl-api-flower/tests/test_abort_run.py index efd272e2c..1b78edac3 100644 --- a/fl-services/flower/fl-api-flower/tests/test_abort_run.py +++ b/fl-services/flower/fl-api-flower/tests/test_abort_run.py @@ -25,7 +25,7 @@ def test_abort_run_success(client, src_root, mock_flwr_run): assert response.status_code == 200 JobMetadata.model_validate(response.json()) - assert response.json() == {"job_id": "9478652229627629048", "status": "STOPPED"} + assert response.json() == {"job_id": "9478652229627629048", "status": "STOPPED", "status_details": None} def test_abort_job_alias_returns_same_shape(client, src_root, mock_flwr_run): @@ -34,7 +34,7 @@ def test_abort_job_alias_returns_same_shape(client, src_root, mock_flwr_run): response = client.delete("/abort_job/9478652229627629048") assert response.status_code == 200 - assert response.json() == {"job_id": "9478652229627629048", "status": "STOPPED"} + assert response.json() == {"job_id": "9478652229627629048", "status": "STOPPED", "status_details": None} def test_abort_run_idempotent_for_terminal_run(client, src_root, mock_flwr_run): @@ -57,7 +57,7 @@ def test_abort_run_idempotent_for_terminal_run(client, src_root, mock_flwr_run): response = client.delete("/abort_run/9478652229627629048") assert response.status_code == 200 - assert response.json() == {"job_id": "9478652229627629048", "status": "FINISHED"} + assert response.json() == {"job_id": "9478652229627629048", "status": "FINISHED", "status_details": None} def test_abort_run_failure_when_run_not_terminal(client, src_root, mock_flwr_run): diff --git a/fl-services/flower/fl-api-flower/tests/test_contract.py b/fl-services/flower/fl-api-flower/tests/test_contract.py index b81dea1cd..0179e8e0d 100644 --- a/fl-services/flower/fl-api-flower/tests/test_contract.py +++ b/fl-services/flower/fl-api-flower/tests/test_contract.py @@ -32,6 +32,7 @@ def test_docs_and_openapi_contract(client): "/submit_run/{job_folder}", "/submit_tutorial/{tutorial_name}", "/abort_run/{run_id}", + "/run_logs/{run_id}", "/upload_app/{model_id}", ): assert path in spec["paths"] @@ -55,7 +56,11 @@ def test_docs_and_openapi_contract(client): assert client_status_schema["items"]["$ref"] == "#/components/schemas/ClientInfoModel" assert list_schema["items"]["$ref"] == "#/components/schemas/JobMetadata" assert submit_schema["type"] == "string" + run_logs_schema = spec["paths"]["/run_logs/{run_id}"]["get"]["responses"]["200"]["content"]["application/json"][ + "schema" + ] assert abort_schema["$ref"] == "#/components/schemas/JobMetadata" + assert run_logs_schema["$ref"] == "#/components/schemas/RunLogs" def test_health_success(client): diff --git a/fl-services/flower/fl-api-flower/tests/test_job_metadata.py b/fl-services/flower/fl-api-flower/tests/test_job_metadata.py index b8b01ed15..8805799c0 100644 --- a/fl-services/flower/fl-api-flower/tests/test_job_metadata.py +++ b/fl-services/flower/fl-api-flower/tests/test_job_metadata.py @@ -13,7 +13,7 @@ import pytest -from fl_api.schemas import JobMetadata, JobStatus, normalize_status +from fl_api.schemas import _FLOWER_STATUS_MAP, JobMetadata, JobStatus, normalize_status @pytest.mark.parametrize( @@ -28,20 +28,44 @@ ("stopped", JobStatus.STOPPED), ("RUNNING", JobStatus.RUNNING), (" running ", JobStatus.RUNNING), - ("", JobStatus.FAILED), + ("", JobStatus.UNKNOWN), ], ) def test_normalize_status_maps_flower_statuses(native, expected): assert normalize_status(native) == expected -def test_normalize_status_unknown_is_failed(): - assert normalize_status("some-future-status") == JobStatus.FAILED +def test_normalize_status_unmapped_is_unknown(): + # UNKNOWN, not FAILED: the hub's failed-job reconcile errors the model and frees the + # net on FAILED, so a status added by a framework upgrade must never be guessed into it. + assert normalize_status("some-future-status") == JobStatus.UNKNOWN -def test_job_metadata_has_exactly_job_id_and_status(): - assert set(JobMetadata.model_fields) == {"job_id", "status"} +def test_normalize_status_covers_every_flwr_status_value(): + """The map is exhaustive against the pinned flwr's own vocabulary. + The UNKNOWN default exists for the statuses a *future* flwr adds; this pins that no + status of the flwr actually installed falls through to it. `flwr ls` reports + non-terminal states bare and terminal states as `finished:` + (see flwr.common.serde / RunStatus), plus the bare `stopped` from `flwr stop`. + """ + from flwr.common.constant import Status, SubStatus -def test_job_status_has_exactly_five_contract_values(): - assert {s.value for s in JobStatus} == {"PENDING", "RUNNING", "FINISHED", "FAILED", "STOPPED"} + non_terminal = [Status.PENDING, Status.STARTING, Status.RUNNING] + terminal = [f"{Status.FINISHED}:{sub}" for sub in (SubStatus.COMPLETED, SubStatus.FAILED, SubStatus.STOPPED)] + for native in non_terminal + terminal: + assert native in _FLOWER_STATUS_MAP, f"flwr status {native!r} is unmapped" + + +def test_job_metadata_has_exactly_the_contract_fields(): + # The other adapter and flip-api's IJobMetaData pin the same set. A field added here and + # nowhere else is invisible until a hub reads it, so all three move in the same PR. + assert set(JobMetadata.model_fields) == {"job_id", "status", "status_details"} + + +def test_status_details_defaults_to_none(): + assert JobMetadata(job_id="1", status=JobStatus.RUNNING).status_details is None + + +def test_job_status_has_exactly_six_contract_values(): + assert {s.value for s in JobStatus} == {"PENDING", "RUNNING", "FINISHED", "FAILED", "STOPPED", "UNKNOWN"} diff --git a/fl-services/flower/fl-api-flower/tests/test_list_runs.py b/fl-services/flower/fl-api-flower/tests/test_list_runs.py index 86e9a36c8..807af3324 100644 --- a/fl-services/flower/fl-api-flower/tests/test_list_runs.py +++ b/fl-services/flower/fl-api-flower/tests/test_list_runs.py @@ -36,8 +36,8 @@ def test_list_runs_success(client, src_root, mock_flwr_run): assert len(body) == 2 for item in body: JobMetadata.model_validate(item) - assert body[0] == {"job_id": "9478652229627629048", "status": "FINISHED"} - assert body[1] == {"job_id": "2528745119497052892", "status": "RUNNING"} + assert body[0] == {"job_id": "9478652229627629048", "status": "FINISHED", "status_details": None} + assert body[1] == {"job_id": "2528745119497052892", "status": "RUNNING", "status_details": None} def test_list_jobs_alias_returns_same_shape(client, src_root, mock_flwr_run): @@ -46,7 +46,69 @@ def test_list_jobs_alias_returns_same_shape(client, src_root, mock_flwr_run): response = client.get("/list_jobs") assert response.status_code == 200 - assert response.json() == [{"job_id": "1", "status": "RUNNING"}] + assert response.json() == [{"job_id": "1", "status": "RUNNING", "status_details": None}] + + +def test_status_details_carries_the_backends_own_cause(client, src_root, mock_flwr_run): + # `flwr ls` already explains a failed run in one line. Carrying it costs nothing — + # no extra call, no log fetch — and it is the cause the hub puts at the top of the + # model's activity feed (FLIP#1001). + mock_flwr_run( + stdout=( + '{"success": true, "runs": [{"run-id":"7","fab-name":"x","status":"finished:failed",' + '"status-details":"ServerApp failed with exception: No module named \'flwr.common.message\'"}]}' + ) + ) + + response = client.get("/list_runs") + + assert response.status_code == 200 + assert response.json() == [ + { + "job_id": "7", + "status": "FAILED", + "status_details": "ServerApp failed with exception: No module named 'flwr.common.message'", + } + ] + + +def test_status_details_na_is_reported_as_absent(client, src_root, mock_flwr_run): + # flwr writes the literal "N/A" for a run with nothing to say rather than omitting the + # key; passing it through would print "Reported cause: N/A" in the activity feed. + mock_flwr_run( + stdout='{"success": true, "runs": [{"run-id":"7","fab-name":"x","status":"running","status-details":"N/A"}]}' + ) + + assert client.get("/list_runs").json() == [{"job_id": "7", "status": "RUNNING", "status_details": None}] + + +def test_status_details_is_collapsed_redacted_and_bounded(client, src_root, mock_flwr_run): + # The text is a researcher-authored exception message from a container holding a hub + # service key, so it gets the same masking the run log does — and a length bound, since + # nothing upstream constrains how long an exception message can be. + secret = "aws_secret_access_key=" + "b" * 900 + mock_flwr_run( + stdout=( + '{"success": true, "runs": [{"run-id":"7","fab-name":"x","status":"finished:failed",' + f'"status-details":"boom\\n over lines {secret}"}}]}}' + ) + ) + + details = client.get("/list_runs").json()[0]["status_details"] + + assert details is not None + assert "\n" not in details + assert "boom over lines" in details + assert "b" * 40 not in details + assert len(details) <= 500 + + +def test_status_details_absent_key_is_not_an_error(client, src_root, mock_flwr_run): + # Older flwr versions have no such key. Unlike run-id/status (which 500 when missing), + # this one is simply absent detail. + mock_flwr_run(stdout='{"success": true, "runs": [{"run-id":"7","fab-name":"x","status":"running"}]}') + + assert client.get("/list_runs").json() == [{"job_id": "7", "status": "RUNNING", "status_details": None}] def test_list_runs_malformed_run_returns_500(client, src_root, mock_flwr_run): diff --git a/fl-services/flower/fl-api-flower/tests/test_redaction.py b/fl-services/flower/fl-api-flower/tests/test_redaction.py new file mode 100644 index 000000000..f7b4bef00 --- /dev/null +++ b/fl-services/flower/fl-api-flower/tests/test_redaction.py @@ -0,0 +1,65 @@ +# Copyright (c) 2026 Guy's and St Thomas' NHS Foundation Trust & King's College London +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# http://www.apache.org/licenses/LICENSE-2.0 +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +import pytest + +from fl_api.utils.redaction import REDACTED, redact_secrets + + +@pytest.mark.parametrize( + ("text", "secret"), + [ + ("X-Internal-Service-Key: abc123def456", "abc123def456"), # pragma: allowlist secret + ("X-Trust-Internal-Service-Key: abc123def456", "abc123def456"), + ("AWS_SECRET_ACCESS_KEY=notARealKeyJustAShape", "notARealKeyJustAShape"), # pragma: allowlist secret + ('{"api_key": "sk-live-9999"}', "sk-live-9999"), + ("password=hunter2", "hunter2"), + ("Bearer token=eyJhbGciOiJIUzI1NiJ9", "eyJhbGciOiJIUzI1NiJ9"), + ("credentials AKIANOTAREALKEYSHAPE in env", "AKIANOTAREALKEYSHAPE"), # pragma: allowlist secret + ], +) +def test_redacts_credential_shapes(text, secret): + redacted = redact_secrets(text) + + assert secret not in redacted + assert REDACTED in redacted + + +def test_redacts_each_sigv4_parameter_without_eating_the_url(): + url = ( + "https://bucket.s3.eu-west-2.amazonaws.com/model/app.py" + "?X-Amz-Credential=AKIANOTAREALKEYSHAPE%2F20260819" # pragma: allowlist secret + "&X-Amz-Signature=deadbeefcafe&X-Amz-Expires=900" + ) + + redacted = redact_secrets(url) + + assert "deadbeefcafe" not in redacted + assert "AKIANOTAREALKEYSHAPE" not in redacted + # The object path and the non-secret parameters stay readable — they are the + # diagnostically useful half of the line. + assert "model/app.py" in redacted + assert "X-Amz-Expires=900" in redacted + + +@pytest.mark.parametrize( + "text", + [ + "ImportError: cannot import name 'min_clients_from_run_config' from 'flip.flower.strategy'", + "KeyError: 'flip-cohort-query'", + "ERROR: Exit Code: 607", + "Traceback (most recent call last):", + ], +) +def test_leaves_diagnostic_text_intact(text): + # Over-redaction is tolerated, but not at the cost of the lines that name the cause. + assert redact_secrets(text) == text diff --git a/fl-services/flower/fl-api-flower/tests/test_run_logs.py b/fl-services/flower/fl-api-flower/tests/test_run_logs.py new file mode 100644 index 000000000..3441cc4ca --- /dev/null +++ b/fl-services/flower/fl-api-flower/tests/test_run_logs.py @@ -0,0 +1,112 @@ +# Copyright (c) 2026 Guy's and St Thomas' NHS Foundation Trust & King's College London +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# http://www.apache.org/licenses/LICENSE-2.0 +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +import subprocess + +from fl_api.schemas import RunLogs + +_SERVERAPP_FAILURE = ( + "INFO: Starting ServerApp\n" + "ERROR: ServerApp raised an exception\n" + "Traceback (most recent call last):\n" + " File 'app/server_app.py', line 26, in \n" + " from flip.flower.strategy import min_clients_from_run_config\n" + "ImportError: cannot import name 'min_clients_from_run_config'\n" + "ERROR: Exit Code: 607\n" +) + + +def test_run_logs_returns_untruncated_log(client, src_root, mock_flwr_run): + mock_flwr_run(stdout=_SERVERAPP_FAILURE) + + response = client.get("/run_logs/9478652229627629048") + + assert response.status_code == 200 + RunLogs.model_validate(response.json()) + assert response.json() == { + "run_id": "9478652229627629048", + "log": _SERVERAPP_FAILURE, + "truncated": False, + } + + +def test_run_logs_uses_show_not_stream(client, src_root, mock_flwr_run): + # The default `--stream` follows the log forever; only `--show` returns. + commands = mock_flwr_run(stdout=_SERVERAPP_FAILURE) + + client.get("/run_logs/9478652229627629048") + + assert commands == [["uvx", "flwr", "log", "9478652229627629048", "local", "--show"]] + + +def test_run_logs_keeps_the_tail_and_flags_truncation(client, src_root, mock_flwr_run, monkeypatch): + # The dependency-install preamble is the half worth dropping; the cause is at the end. + monkeypatch.setenv("FLOWER_RUN_LOG_MAX_CHARS", "120") + preamble = "".join(f"Installed package-{index}\n" for index in range(200)) + mock_flwr_run(stdout=preamble + _SERVERAPP_FAILURE) + + response = client.get("/run_logs/1") + + body = response.json() + assert response.status_code == 200 + assert body["truncated"] is True + assert "ERROR: Exit Code: 607" in body["log"] + assert "Installed package-0" not in body["log"] + # The cut is advanced to a line boundary, so the tail never opens mid-line. + assert not body["log"].startswith("nstalled") + assert len(body["log"]) <= 120 + + +def test_run_logs_redacts_credentials(client, src_root, mock_flwr_run): + mock_flwr_run(stdout="X-Internal-Service-Key: s3cr3t-value\nERROR: Exit Code: 607\n") + + response = client.get("/run_logs/1") + + assert response.status_code == 200 + assert "s3cr3t-value" not in response.json()["log"] + assert "ERROR: Exit Code: 607" in response.json()["log"] + + +def test_run_logs_returns_500_when_flwr_fails(client, src_root, mock_flwr_run): + mock_flwr_run(returncode=1, stderr="Invalid run_id `1`, exiting") + + response = client.get("/run_logs/1") + + assert response.status_code == 500 + assert "Invalid run_id" in response.json()["detail"] + + +def test_run_logs_returns_500_when_flwr_times_out(client, src_root, mock_flwr_run): + # A wedged CLI (unreachable SuperLink) must fail the request, not hang the caller. + mock_flwr_run(exception=subprocess.TimeoutExpired(cmd="flwr log", timeout=60)) + + response = client.get("/run_logs/1") + + assert response.status_code == 500 + + +def test_run_logs_rejects_non_numeric_run_id(client, src_root): + # Same guard as /abort_run: a non-numeric segment never reaches the `flwr` argv. + response = client.get("/run_logs/not-a-number") + + assert response.status_code == 422 + + +def test_run_logs_invalid_max_chars_falls_back_to_default(client, src_root, mock_flwr_run, monkeypatch): + # A broken operator value must not truncate to nothing (or blow up the request). + monkeypatch.setenv("FLOWER_RUN_LOG_MAX_CHARS", "not-a-number") + mock_flwr_run(stdout=_SERVERAPP_FAILURE) + + response = client.get("/run_logs/1") + + assert response.status_code == 200 + assert response.json()["log"] == _SERVERAPP_FAILURE diff --git a/fl-services/nvflare/fl-api-base/fl_api/utils/schemas.py b/fl-services/nvflare/fl-api-base/fl_api/utils/schemas.py index 3f881e978..618df3df2 100644 --- a/fl-services/nvflare/fl-api-base/fl_api/utils/schemas.py +++ b/fl-services/nvflare/fl-api-base/fl_api/utils/schemas.py @@ -145,6 +145,11 @@ class JobStatus(StrEnum): FINISHED = "FINISHED" FAILED = "FAILED" STOPPED = "STOPPED" + # A native status the map below does not recognise — i.e. a framework upgrade added one. + # Never guessed into a real state: FAILED now has destructive consumers on the hub (the + # failed-job reconcile errors the model and frees the net), and RUNNING would make the + # job abortable, so an unmapped status must reach the hub as explicitly unknowable. + UNKNOWN = "UNKNOWN" class JobMetadata(BaseModel): @@ -152,6 +157,14 @@ class JobMetadata(BaseModel): job_id: str status: JobStatus + # Part of the shared contract, but always None here: NVFLARE has no native per-job + # explanation to carry. `nvflare.apis.job_def.JobMetaKey` exposes no error/details key — + # the closest are `job_deploy_detail` (per-target "OK"/failure of the *deployment* only; + # it reads `['server: OK', 'Trust_2: OK']` even on a job that then died with + # FINISHED:EXECUTION_EXCEPTION) and `schedule_history`. The cause of an NVFLARE run + # failure lives only in the fl-server container's stdout. Declared so the field means the + # same thing on both adapters and the hub never has to branch on backend to read it. + status_details: str | None = None # NVFLARE RunStatus (nvflare/apis/job_def.py) -> normalized contract status. @@ -178,10 +191,11 @@ def normalize_status(native_status: str) -> JobStatus: Returns: JobStatus: The normalized status. Unknown / unmapped statuses are logged and - treated as ``FAILED`` — never silently surfaced as an abortable ``RUNNING``. + surfaced as ``UNKNOWN`` — never guessed into an abortable ``RUNNING`` or an + actionable ``FAILED`` (the hub's failed-job reconcile acts on FAILED). """ normalized = _NVFLARE_STATUS_MAP.get(native_status.strip().upper()) if normalized is None: - logger.warning("Unmapped NVFLARE job status %r; treating as FAILED.", native_status) - return JobStatus.FAILED + logger.warning("Unmapped NVFLARE job status %r; surfacing as UNKNOWN.", native_status) + return JobStatus.UNKNOWN return normalized diff --git a/fl-services/nvflare/fl-api-base/tests/routers/test_jobs.py b/fl-services/nvflare/fl-api-base/tests/routers/test_jobs.py index e796d13cd..896179715 100644 --- a/fl-services/nvflare/fl-api-base/tests/routers/test_jobs.py +++ b/fl-services/nvflare/fl-api-base/tests/routers/test_jobs.py @@ -70,8 +70,10 @@ def test_list_jobs_success(client): jobs = response.json() assert isinstance(jobs, list) assert jobs == [ - {"job_id": "1234", "status": "FINISHED"}, - {"job_id": "5678", "status": "RUNNING"}, + # status_details is part of the shared contract but always None here: NVFLARE has + # no native per-job explanation to carry (see JobMetadata in utils/schemas.py). + {"job_id": "1234", "status": "FINISHED", "status_details": None}, + {"job_id": "5678", "status": "RUNNING", "status_details": None}, ] @@ -108,7 +110,7 @@ def test_reset_errors_job_not_found(client): def test_abort_job_success(client): response = client.delete("/abort_job/1234") assert response.status_code == status.HTTP_200_OK - assert response.json() == {"job_id": "1234", "status": "STOPPED"} + assert response.json() == {"job_id": "1234", "status": "STOPPED", "status_details": None} def test_abort_job_not_found(client): diff --git a/fl-services/nvflare/fl-api-base/tests/test_job_metadata.py b/fl-services/nvflare/fl-api-base/tests/test_job_metadata.py index 1e33ddb1a..84f5d86fa 100644 --- a/fl-services/nvflare/fl-api-base/tests/test_job_metadata.py +++ b/fl-services/nvflare/fl-api-base/tests/test_job_metadata.py @@ -31,15 +31,17 @@ ("FINISHED:ABANDONED", JobStatus.FAILED), ("running", JobStatus.RUNNING), (" RUNNING ", JobStatus.RUNNING), - ("", JobStatus.FAILED), + ("", JobStatus.UNKNOWN), ], ) def test_normalize_status_maps_nvflare_runstatus(native, expected): assert normalize_status(native) == expected -def test_normalize_status_unknown_is_failed(): - assert normalize_status("some-future-status") == JobStatus.FAILED +def test_normalize_status_unmapped_is_unknown(): + # UNKNOWN, not FAILED: the hub's failed-job reconcile errors the model and frees the + # net on FAILED, so a status added by a framework upgrade must never be guessed into it. + assert normalize_status("some-future-status") == JobStatus.UNKNOWN def test_normalize_status_covers_every_runstatus_value(): @@ -49,9 +51,15 @@ def test_normalize_status_covers_every_runstatus_value(): assert run_status.value in _NVFLARE_STATUS_MAP, f"RunStatus.{run_status.name} is unmapped" -def test_job_metadata_has_exactly_job_id_and_status(): - assert set(JobMetadata.model_fields) == {"job_id", "status"} +def test_job_metadata_has_exactly_the_contract_fields(): + # The other adapter and flip-api's IJobMetaData pin the same set. A field added here and + # nowhere else is invisible until a hub reads it, so all three move in the same PR. + assert set(JobMetadata.model_fields) == {"job_id", "status", "status_details"} -def test_job_status_has_exactly_five_contract_values(): - assert {s.value for s in JobStatus} == {"PENDING", "RUNNING", "FINISHED", "FAILED", "STOPPED"} +def test_status_details_defaults_to_none(): + assert JobMetadata(job_id="1", status=JobStatus.RUNNING).status_details is None + + +def test_job_status_has_exactly_six_contract_values(): + assert {s.value for s in JobStatus} == {"PENDING", "RUNNING", "FINISHED", "FAILED", "STOPPED", "UNKNOWN"} diff --git a/flip-api/src/flip_api/config.py b/flip-api/src/flip_api/config.py index c84b50c33..d49c14d71 100644 --- a/flip-api/src/flip_api/config.py +++ b/flip-api/src/flip_api/config.py @@ -129,6 +129,15 @@ class Settings(BaseSettings): 30 # How often to check for projects with unimported studies (in minutes) ) SCHEDULER_MALWARE_SCAN_RECONCILE_RATE: int = 1 # How often to reconcile stuck SCANNING uploads (in minutes) + # How often to ask each net's FL API whether an in-flight job has failed (in minutes). + # Bounds how long a run that dies after submission can leave its model looking alive. + SCHEDULER_FL_JOB_RECONCILE_RATE: int = 1 + # How long (in minutes) an in-flight job may go unlisted by its FL backend before the + # reconcile treats it as dead. Covers a backend restart losing its run state (the + # SuperLink's is in-memory by default): the run can never report, so waiting longer + # just leaves the model looking alive. Generous so a transient listing hiccup never + # errors a healthy run. + FL_JOB_UNLISTED_GRACE_MINUTES: int = 30 # Database settings DB_PORT: int @@ -218,11 +227,13 @@ def coerce_empty_pre_signed_url_expiration(cls, v: object) -> object: "PICKLESCAN_TIMEOUT_SECONDS", "BANDIT_TIMEOUT_SECONDS", "SCHEDULER_MALWARE_SCAN_RECONCILE_RATE", + "SCHEDULER_FL_JOB_RECONCILE_RATE", + "FL_JOB_UNLISTED_GRACE_MINUTES", mode="before", ) @classmethod - def coerce_empty_scan_int(cls, v: object, info: ValidationInfo) -> object: - """Treat an empty-string scan-timing setting as the field default. + def coerce_empty_interval_int(cls, v: object, info: ValidationInfo) -> object: + """Treat an empty-string sweep-timing setting as the field default. Same rationale as ``coerce_empty_max_model_file_bytes``: these arrive as empty strings whenever the name appears in an env file at all — diff --git a/flip-api/src/flip_api/domain/interfaces/fl.py b/flip-api/src/flip_api/domain/interfaces/fl.py index 2589b454f..f9ce1fa47 100644 --- a/flip-api/src/flip_api/domain/interfaces/fl.py +++ b/flip-api/src/flip_api/domain/interfaces/fl.py @@ -104,7 +104,7 @@ class IJobMetaData(BaseModel): The shared job-metadata contract (GitHub issue #490). flip-api correlates ``model_id`` <-> ``job_id`` in its own ``fl_job`` table, so the contract carries - only ``job_id`` + ``status``. + ``job_id`` + ``status``, plus the optional ``status_details`` one-liner below. ``job_id`` is the backend-assigned identifier, treated as an opaque string the hub never parses: a UUID-like string for NVFLARE, a stringified integer run-id for Flower. It is @@ -115,6 +115,13 @@ class IJobMetaData(BaseModel): job_id: str status: FLJobStatus + # The backend's own one-line explanation of ``status``, when it has one. Optional and + # defaulted so an FL API predating the field still validates -- and so the hub reads it + # the same way on both backends: Flower fills it from `flwr ls`'s `status-details` (for a + # failed run, the ServerApp's exception type and message), NVFLARE has no equivalent + # native field and always leaves it None. Never load-bearing: it is diagnostic text for + # the activity feed, never an input to a status decision. + status_details: str | None = None class IRequiredTrainingInformation(BaseModel): diff --git a/flip-api/src/flip_api/domain/schemas/status.py b/flip-api/src/flip_api/domain/schemas/status.py index dfc03ed9c..0d84efa90 100644 --- a/flip-api/src/flip_api/domain/schemas/status.py +++ b/flip-api/src/flip_api/domain/schemas/status.py @@ -211,3 +211,8 @@ class FLJobStatus(StrEnum): FINISHED = "FINISHED" FAILED = "FAILED" STOPPED = "STOPPED" + # A native status the adapter's map does not recognise (i.e. a framework upgrade added + # one). Deliberately distinct from FAILED: consumers that act on FAILED are destructive + # (the failed-job reconcile errors the model and frees the net), so a guess must never + # act — UNKNOWN is a no-op everywhere, surfaced only in the adapter's warning log. + UNKNOWN = "UNKNOWN" diff --git a/flip-api/src/flip_api/fl_services/reconcile_failed_jobs.py b/flip-api/src/flip_api/fl_services/reconcile_failed_jobs.py new file mode 100644 index 000000000..d3da6ed42 --- /dev/null +++ b/flip-api/src/flip_api/fl_services/reconcile_failed_jobs.py @@ -0,0 +1,283 @@ +# Copyright (c) Guy's and St Thomas' NHS Foundation Trust & King's College London +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# http://www.apache.org/licenses/LICENSE-2.0 +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +"""Surface FL runs that fail after the hub has already submitted them (FLIP#1001). + +``start_training`` submits a job, stores the backend's job id on ``FLJob`` and never asks +about it again. Everything the hub knows about a run after that arrives *from* the run — +the ServerApp reports rounds, metrics and its final status through the ``flip`` package. +A run that dies before any of that happens (an ImportError at ServerApp module scope is +the canonical case) therefore reports nothing at all, and the model sits at ``INITIATED`` +forever with the cause visible only to whoever knows to run ``flwr log`` inside the FL API +container. + +This sweep closes that gap from the one side that can: the hub already holds the job id, +the net the job is pinned to, and the authority to move the model. Each tick it asks every +in-flight job's FL API for that job's status and, on ``FAILED``, records the backend's log +tail against the model and drives it to ``ERROR`` — which releases the net through +``update_model_status``'s existing terminal-status path. + +That same ``/list_jobs`` response also carries ``status_details`` — the backend's own one-line +explanation, where it has one — so the feed row can lead with the cause rather than opening on +the per-run dependency install that dominates a Flower log tail. Flower fills it; NVFLARE has +no equivalent native field and leaves it unset. + +Two conditions act, both meaning "this run will never report": + +* the backend lists the job as ``FAILED``, or +* the backend does not list the job at all and it was submitted more than + ``FL_JOB_UNLISTED_GRACE_MINUTES`` ago. A SuperLink keeps run state in memory by default, + so a restart forgets every run — the exact silent-death this sweep exists for, just with + the evidence gone too. The grace period keeps a transient listing hiccup from erroring a + healthy run. + +Deliberately narrow beyond that, because the sweep's own errors would be much worse than +the bug it fixes: + +* ``FINISHED`` is left alone — a run whose ServerApp has finished is routinely still + uploading results, and the hub's own ``RESULTS_UPLOADED`` callback is the authority on + that. ``UNKNOWN`` (a native status the adapter's map does not recognise) is left alone — + it is exactly the case where acting would be guessing. +* only jobs the hub still considers in flight, whose model has not already reached a + terminal status, are polled at all; the model status is re-read after the (network) status + call so a result landing mid-poll wins the race. +* everything is per-job best-effort: an unreachable FL API, a malformed response or a failed + write is logged and the sweep moves to the next job. + +Scope: this covers the **ServerApp**. A ClientApp that dies at a trust logs to that trust's +SuperNode, which the hub cannot read, and is out of scope here. +""" + +from datetime import datetime, timedelta +from typing import cast +from uuid import UUID + +from sqlmodel import Session, col, select + +from flip_api.config import get_settings +from flip_api.db.database import get_engine +from flip_api.db.models.main_models import FLJob, FLNets, FLScheduler, Model +from flip_api.domain.schemas.status import FLJobStatus, JobStatus, ModelStatus +from flip_api.domain.schemas.types import FLBackend +from flip_api.fl_services.services.fl_service import fetch_run_logs, get_backend_job_metadata +from flip_api.model_services.services.model_service import add_log, update_model_status +from flip_api.utils.logger import logger + +# Model statuses that mean "the hub is still waiting on this run". A model outside this set +# has already been resolved (by the run itself, by an abort, or by an earlier sweep), so its +# job is not polled — which also makes the sweep idempotent: the ERROR it writes takes the +# model out of the set. +_IN_FLIGHT_MODEL_STATUSES = (ModelStatus.INITIATED, ModelStatus.PREPARED, ModelStatus.RUNNING) + +# Second cap on the stored log tail. fl-api-flower already truncates before returning, so +# this only guards against an FL API that doesn't — a runaway body must not become a +# runaway `fl_logs` row. +_MAX_STORED_LOG_CHARS = 8000 + +# Where to look when the sweep could not produce the log itself, per backend. Only +# fl-api-flower serves /run_logs today, so every NVFLARE failure lands on its hint. +_MANUAL_FALLBACK_HINTS: dict[FLBackend, str] = { + FLBackend.FLOWER: ( + "The FL run log could not be retrieved from the FL API. Run " + "`flwr log local --show` inside the net's FL API container to read it." + ), + # Deliberately points at the container's stdout and nowhere else. The two more obvious + # destinations do not work: the FL API's `POST //show_errors/server` 500s against + # the installed NVFLARE (FLIP#1032), and the job workspace holds opaque + # `data`/`meta`/`workspace` blobs rather than a readable log — a filesystem-wide search of + # a failed run's fl-server matched only the uploaded app bundle, never the traceback. + # Revisit once #1032 lands. + FLBackend.NVFLARE: ( + "The FL run log could not be retrieved from the FL API. Read the failure in the net's " + "fl-server container output (`docker logs fl-server-net-`); NVFLARE keeps no " + "readable per-job log in the job workspace." + ), +} + + +def _failure_message( + fl_backend_job_id: str, status_details: str | None, run_log: str | None, fl_backend: FLBackend +) -> str: + """Compose the activity-feed text for a run that failed after submission. + + ``status_details`` leads when the backend supplied one, because the log tail buries the + cause: a Flower run log opens with the per-run ``uv sync``, whose single ``Installed: + [...]`` line is most of the stored row, leaving the traceback below the fold of the feed + panel. The one-liner is the same exception the traceback ends with, so putting it at the + top costs a line and saves the reader the scroll. The tail still follows — it carries the + file and line number, which the one-liner does not. + + Args: + fl_backend_job_id (str): The backend-assigned job id, quoted so an operator can + take it straight to ``flwr log`` / the FL API. + status_details (str | None): The backend's own one-line cause, when it has one. + run_log (str | None): The backend's log tail, when it could be retrieved. + fl_backend (FLBackend): The net's backend, selecting the manual-fallback wording. + + Returns: + str: The text stored on the model's failed ``fl_logs`` row. + """ + header = f"Training failed: FL run {fl_backend_job_id} ended in a failed state without reporting a result." + if status_details: + header = f"{header}\n\nReported cause: {status_details}" + if not run_log or not run_log.strip(): + return f"{header}\n\n{_MANUAL_FALLBACK_HINTS[fl_backend]}" + return f"{header}\n\nEnd of the FL run log:\n{run_log.strip()[-_MAX_STORED_LOG_CHARS:]}" + + +def _unlisted_message(fl_backend_job_id: str, started: datetime) -> str: + """Compose the activity-feed text for a run its own backend no longer lists. + + No log tail is offered: the backend that has forgotten the run (a SuperLink restart + loses its in-memory run state) has forgotten its log with it. + + Args: + fl_backend_job_id (str): The backend-assigned job id of the vanished run. + started (datetime): When the hub started the job (naive UTC, ``FLJob.started``). + + Returns: + str: The text stored on the model's failed ``fl_logs`` row. + """ + return ( + f"Training failed: FL run {fl_backend_job_id} (started {started:%Y-%m-%d %H:%M} UTC) is no " + "longer listed by its FL backend and never reported a result. The backend has likely " + "restarted since submission, losing the run and its log." + ) + + +def _resolve_failure(model_id: UUID, fl_backend_job_id: str, message: str, session: Session) -> None: + """Record a dead run against its model and move the model to ``ERROR``. + + The log row is added before the status change so the activity feed reads + cause-then-verdict, matching the prepare-failure path in ``prepare_and_start_training`` + — but with ``transaction=session`` so it commits *with* ``update_model_status``'s own + commit rather than on its own: a status write that fails must not leave an orphaned + feed row for the next tick to duplicate. ``update_model_status(ERROR)`` completes the + FL job and frees the net. + + Args: + model_id (UUID): The model whose run died. + fl_backend_job_id (str): The backend-assigned job id of the dead run. + message (str): The activity-feed text explaining the failure. + session (Session): SQLModel session. + + Returns: + None + """ + add_log(model_id, message, session, transaction=session, success=False) + update_model_status(model_id, ModelStatus.ERROR, session) + logger.error(f"FL run {fl_backend_job_id} for model {model_id} failed; model set to ERROR.") + + +def reconcile_failed_fl_jobs(session: Session) -> int: + """Check every in-flight FL job with the backend and resolve the ones that failed. + + Args: + session (Session): SQLModel session. + + Returns: + int: The number of models moved to ``ERROR`` by this pass. + """ + # One row per in-flight job, joined through its *own* scheduler so a model that has + # been retried is checked against the net its current job is pinned to rather than + # whichever net a stale job once used. + statement = ( + # SQLModel's typed select() overloads stop at four entities; the same ignore is + # used for the six-column select in model_service._run_trusts_by_model. + select( # type: ignore[call-overload] + FLJob.id, + FLJob.model_id, + FLJob.fl_backend_job_id, + FLJob.started, + FLNets.endpoint, + FLNets.fl_backend, + ) + .join(FLScheduler, col(FLScheduler.job_id) == col(FLJob.id)) + .join(FLNets, col(FLNets.id) == col(FLScheduler.net_id)) + .join(Model, col(Model.id) == col(FLJob.model_id)) + .where( + FLJob.status == JobStatus.IN_PROGRESS, + col(FLJob.fl_backend_job_id).is_not(None), + col(Model.status).in_(_IN_FLIGHT_MODEL_STATUSES), + ) + ) + jobs = session.exec(statement).all() + if not jobs: + return 0 + + unlisted_grace = timedelta(minutes=get_settings().FL_JOB_UNLISTED_GRACE_MINUTES) + reported = 0 + for job_id, model_id, nullable_backend_job_id, started, endpoint, fl_backend in jobs: + # Non-null by the WHERE clause above; the column is nullable until submission. + fl_backend_job_id = cast(str, nullable_backend_job_id) + try: + backend_job = get_backend_job_metadata(endpoint, fl_backend_job_id) + backend_status = backend_job.status if backend_job else None + + if backend_job is not None and backend_status == FLJobStatus.FAILED: + # The log fetch goes over the network too, so do it before the race + # re-check below. + run_log = fetch_run_logs(endpoint, fl_backend_job_id) + message = _failure_message(fl_backend_job_id, backend_job.status_details, run_log, fl_backend) + elif backend_status is None and started is not None and datetime.utcnow() - started > unlisted_grace: + # The backend has no record of a run the hub submitted long ago — a + # backend restart lost it (SuperLink run state is in-memory by default). + # It can never report now, so it gets the same resolution as FAILED, + # minus the unfetchable log. Comfortably past any submission race: + # `started` predates the submit that minted fl_backend_job_id. + message = _unlisted_message(fl_backend_job_id, started) + else: + # PENDING / RUNNING: alive. FINISHED: the run's own RESULTS_UPLOADED + # callback is the authority. STOPPED: the abort path already resolved it. + # UNKNOWN: acting on an unrecognised status would be guessing. + # None within the grace period: a listing hiccup, retried next tick. + continue + + # The calls above went over the network; a result or an abort can have landed + # in the meantime. Re-read the model's status (a column select, so it bypasses + # the session identity map and sees the latest committed value) and defer to + # whatever won. + current_status = session.exec(select(col(Model.status)).where(Model.id == model_id)).one_or_none() + if current_status not in _IN_FLIGHT_MODEL_STATUSES: + logger.info( + f"FL run {fl_backend_job_id} is dead ({backend_status or 'unlisted'}) but model " + f"{model_id} has already settled as {current_status}; leaving it alone." + ) + continue + + _resolve_failure(model_id, fl_backend_job_id, message, session) + reported += 1 + except Exception as e: + # One unreachable net (or one poisoned write) must not stop the other jobs being + # checked. Roll back first: a failed write leaves the transaction unusable, and + # every subsequent row would raise on it. + session.rollback() + logger.error( + f"Failed to reconcile FL job {job_id} (backend id {fl_backend_job_id}): {type(e).__name__}: {e}" + ) + + return reported + + +def reconcile_failed_fl_jobs_scheduled_task() -> None: + """Scheduled entry point for :func:`reconcile_failed_fl_jobs`. + + Never raises: the sweep is a safety net, and a failure in it must not take down the + background scheduler. + """ + try: + with Session(get_engine()) as db: + reported = reconcile_failed_fl_jobs(db) + if reported: + logger.info(f"FL job reconcile marked {reported} model(s) as errored.") + except Exception as e: + logger.error(f"Error in scheduled FL job reconcile: {type(e).__name__}: {e}") diff --git a/flip-api/src/flip_api/fl_services/services/fl_service.py b/flip-api/src/flip_api/fl_services/services/fl_service.py index 94c52f987..434ac076a 100644 --- a/flip-api/src/flip_api/fl_services/services/fl_service.py +++ b/flip-api/src/flip_api/fl_services/services/fl_service.py @@ -977,6 +977,76 @@ def extract_current_job_data(net_endpoint: str, fl_backend_job_id: str) -> IJobM return current_job_data[0] +def get_backend_job_metadata(net_endpoint: str, fl_backend_job_id: str) -> IJobMetaData | None: + """Return the FL backend's own metadata for one submitted job. + + Unlike ``extract_current_job_data`` this does not filter to running jobs — the point + is to see terminal states, in particular the ``FAILED`` a run reaches when it dies + after a successful submission (FLIP#1001). + + Returns the whole contract item rather than just the status so the caller also gets + ``status_details`` — the backend's own one-line cause, free in this same response — + without a second network call. + + Args: + net_endpoint (str): The endpoint of the FL API service. + fl_backend_job_id (str): The backend-assigned job id to look up. + + Returns: + IJobMetaData | None: The job's metadata, or ``None`` when the backend does not list + the job at all (nothing can be concluded — callers must not treat that as a + failure). + + Raises: + ValueError: If the FL server response is not a list. + pydantic.ValidationError: If an entry does not conform to ``IJobMetaData``. + """ + url = f"{net_endpoint}/list_jobs" + # Same generous timeout as the other FL status checks: listing runs goes through to the + # SuperLink / FLARE admin session and can exceed httpx's 5s default. + response = http_get(url, timeout=30) + + if not isinstance(response, list): + error_msg = f"Unexpected response format from {url}: {response}" + logger.error(error_msg) + raise ValueError(error_msg) + + for job in (IJobMetaData.model_validate(entry) for entry in response): + if job.job_id == fl_backend_job_id: + return job + + logger.info(f"Job {fl_backend_job_id} is not listed by the FL API at {net_endpoint}.") + return None + + +def fetch_run_logs(net_endpoint: str, fl_backend_job_id: str) -> str | None: + """Fetch the FL backend's log tail for one job, best-effort. + + Only fl-api-flower serves ``/run_logs`` today, and it is diagnostic detail rather than + control flow, so every failure mode — endpoint absent, FL API down, malformed body — + degrades to ``None`` and the caller reports the failure without it. + + Args: + net_endpoint (str): The endpoint of the FL API service. + fl_backend_job_id (str): The backend-assigned job id whose log to fetch. + + Returns: + str | None: The log tail, or ``None`` when it could not be retrieved. + """ + url = f"{net_endpoint}/run_logs/{fl_backend_job_id}" + try: + response = http_get(url, timeout=60) + except Exception as e: + logger.warning(f"Could not fetch run logs from {url}: {type(e).__name__}: {e}") + return None + + if not isinstance(response, dict) or not isinstance(response.get("log"), str): + logger.warning(f"Unexpected run-logs response format from {url}: {type(response).__name__}") + return None + + return cast(str, response["log"]) + + def abort_model_training(request: Request, model_id: UUID, session: Session) -> None: """ Check if the model is currently running training, and if it is, send an abort request to the FL server. diff --git a/flip-api/src/flip_api/scheduler/apscheduler_runner.py b/flip-api/src/flip_api/scheduler/apscheduler_runner.py index db8cbc54f..ae3ee486c 100644 --- a/flip-api/src/flip_api/scheduler/apscheduler_runner.py +++ b/flip-api/src/flip_api/scheduler/apscheduler_runner.py @@ -14,6 +14,7 @@ from flip_api.config import get_settings from flip_api.file_services.services.malware_scan_service import reconcile_scanning_files_scheduled_task +from flip_api.fl_services.reconcile_failed_jobs import reconcile_failed_fl_jobs_scheduled_task from flip_api.fl_services.run_jobs import run_jobs_scheduled_task from flip_api.fl_services.services.fl_service import keep_fl_api_session_alive from flip_api.private_services.stale_task_recovery import recover_stale_tasks_scheduled_task @@ -51,6 +52,11 @@ "interval", minutes=get_settings().SCHEDULER_MALWARE_SCAN_RECONCILE_RATE, ) +scheduler.add_job( + reconcile_failed_fl_jobs_scheduled_task, + "interval", + minutes=get_settings().SCHEDULER_FL_JOB_RECONCILE_RATE, +) def start_scheduler() -> None: diff --git a/flip-api/tests/integration/test_fl_reconcile_db_flow.py b/flip-api/tests/integration/test_fl_reconcile_db_flow.py new file mode 100644 index 000000000..1e8d9bb6c --- /dev/null +++ b/flip-api/tests/integration/test_fl_reconcile_db_flow.py @@ -0,0 +1,149 @@ +# Copyright (c) Guy's and St Thomas' NHS Foundation Trust & King's College London +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# http://www.apache.org/licenses/LICENSE-2.0 +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +"""Integration coverage of the failed-FL-job reconcile (#1001) against the throwaway Postgres. + +Two things here are SQL-shaped and silently passable under mocked sessions: the +FLJob→FLScheduler→FLNets→Model join that selects the in-flight jobs (a wrong join returns +zero rows and the sweep becomes a no-op nobody notices), and the cross-table resolution the +sweep delegates to ``update_model_status(ERROR)`` — the FLJob completing and the scheduler +returning to AVAILABLE. +""" + +from datetime import datetime, timedelta +from unittest.mock import patch +from uuid import uuid4 + +import pytest +from sqlmodel import select + +from flip_api.db.models.main_models import FLJob, FLLogs, FLNets, FLScheduler +from flip_api.domain.interfaces.fl import IJobMetaData +from flip_api.domain.schemas.status import FLJobStatus, JobStatus, ModelStatus, NetStatus, ProjectStatus +from flip_api.domain.schemas.types import FLBackend +from flip_api.fl_services.reconcile_failed_jobs import reconcile_failed_fl_jobs + +_BACKEND_JOB_ID = "11536428743664681318" + + +@pytest.fixture +def submitted_in_flight_job(session, user_factory, project_factory, model_factory): + """A submitted, in-flight job: model INITIATED, job IN_PROGRESS with a backend id, net BUSY. + + Mirrors the state after ``submit_job`` returned and before the run reports anything — + the window the reconcile exists for. + """ + user = user_factory() + project = project_factory.build(owner_id=user.id, status=ProjectStatus.APPROVED, deleted=False) + session.add(project) + session.flush() + + model = model_factory.build( + project_id=project.id, + owner_id=user.id, + status=ModelStatus.INITIATED, + deleted=False, + ) + session.add(model) + session.flush() + + net = FLNets(name=f"net-{uuid4()}", endpoint=f"http://fl-api-{uuid4()}:5000", fl_backend=FLBackend.FLOWER) + session.add(net) + session.flush() + + job = FLJob( + model_id=model.id, + status=JobStatus.IN_PROGRESS, + started=datetime.utcnow() - timedelta(minutes=5), + fl_backend_job_id=_BACKEND_JOB_ID, + ) + session.add(job) + session.flush() + + scheduler = FLScheduler(net_id=net.id, status=NetStatus.BUSY, job_id=job.id) + session.add(scheduler) + session.commit() + + return {"model": model, "job": job, "scheduler": scheduler, "net": net} + + +def test_failed_run_is_resolved_end_to_end(session, submitted_in_flight_job): + """A FAILED run errors the model, records the log tail, completes the job and frees the net.""" + ctx = submitted_in_flight_job + + with ( + patch( + "flip_api.fl_services.reconcile_failed_jobs.get_backend_job_metadata", + return_value=IJobMetaData(job_id=_BACKEND_JOB_ID, status=FLJobStatus.FAILED), + ) as mock_status, + patch( + "flip_api.fl_services.reconcile_failed_jobs.fetch_run_logs", + return_value="ERROR: ServerApp raised an exception\nImportError: cannot import name 'x'", + ), + ): + reported = reconcile_failed_fl_jobs(session) + + assert reported == 1 + mock_status.assert_called_once_with(ctx["net"].endpoint, _BACKEND_JOB_ID) + + session.expire_all() + assert ctx["model"].status == ModelStatus.ERROR + job = session.get(FLJob, ctx["job"].id) + scheduler = session.get(FLScheduler, ctx["scheduler"].id) + assert job is not None + assert job.status == JobStatus.COMPLETED + assert scheduler is not None + assert scheduler.status == NetStatus.AVAILABLE + + logs = session.exec(select(FLLogs).where(FLLogs.model_id == ctx["model"].id)).all() + failure_rows = [row for row in logs if row.success is False] + assert len(failure_rows) == 1 + assert _BACKEND_JOB_ID in (failure_rows[0].log or "") + assert "ImportError" in (failure_rows[0].log or "") + + +def test_settled_job_is_not_even_polled(session, submitted_in_flight_job): + """A COMPLETED job (or settled model) never reaches the FL API — the join filters it out.""" + ctx = submitted_in_flight_job + ctx["job"].status = JobStatus.COMPLETED + session.add(ctx["job"]) + session.commit() + + with patch("flip_api.fl_services.reconcile_failed_jobs.get_backend_job_metadata") as mock_status: + reported = reconcile_failed_fl_jobs(session) + + assert reported == 0 + mock_status.assert_not_called() + + +def test_unlisted_run_past_grace_is_resolved_end_to_end(session, submitted_in_flight_job): + """A run its backend forgot (SuperLink restart) is resolved once past the grace period.""" + ctx = submitted_in_flight_job + ctx["job"].started = datetime.utcnow() - timedelta(hours=2) + session.add(ctx["job"]) + session.commit() + + with ( + patch("flip_api.fl_services.reconcile_failed_jobs.get_backend_job_metadata", return_value=None), + patch("flip_api.fl_services.reconcile_failed_jobs.fetch_run_logs") as mock_logs, + ): + reported = reconcile_failed_fl_jobs(session) + + assert reported == 1 + mock_logs.assert_not_called() + + session.expire_all() + assert ctx["model"].status == ModelStatus.ERROR + logs = session.exec(select(FLLogs).where(FLLogs.model_id == ctx["model"].id)).all() + failure_rows = [row for row in logs if row.success is False] + assert len(failure_rows) == 1 + assert "no longer listed" in (failure_rows[0].log or "") diff --git a/flip-api/tests/unit/fl_services/services/test_fl_service.py b/flip-api/tests/unit/fl_services/services/test_fl_service.py index ab4dcf279..5e27b9608 100644 --- a/flip-api/tests/unit/fl_services/services/test_fl_service.py +++ b/flip-api/tests/unit/fl_services/services/test_fl_service.py @@ -25,7 +25,7 @@ IServerStatus, IStartTrainingBody, ) -from flip_api.domain.schemas.status import ClientStatus, JobStatus +from flip_api.domain.schemas.status import ClientStatus, FLJobStatus, JobStatus from flip_api.domain.schemas.types import FLBackend from flip_api.fl_services.services import fl_service from flip_api.utils.exceptions import DatabaseError, JobAbortedError, NotFoundError @@ -1090,6 +1090,113 @@ def test_extract_current_job_data_multiple_found(mock_http_get): extract_current_job_data(net_endpoint, fl_backend_job_id) +@patch("flip_api.fl_services.services.fl_service.http_get") +def test_get_backend_job_metadata_returns_terminal_status(mock_http_get): + # Unlike extract_current_job_data this must see terminal states — that is the whole + # point of the FL job reconcile (#1001). + from flip_api.fl_services.services.fl_service import get_backend_job_metadata + + mock_http_get.return_value = [ + {"job_id": "job999", "status": "RUNNING"}, + {"job_id": "job123", "status": "FAILED"}, + ] + + assert get_backend_job_metadata("http://fl-api-endpoint", "job123").status == FLJobStatus.FAILED + mock_http_get.assert_called_once_with("http://fl-api-endpoint/list_jobs", timeout=30) + + +@patch("flip_api.fl_services.services.fl_service.http_get") +def test_get_backend_job_metadata_unlisted_job_returns_none(mock_http_get): + # "Not listed" is not "failed": the caller must be able to tell them apart. + from flip_api.fl_services.services.fl_service import get_backend_job_metadata + + mock_http_get.return_value = [{"job_id": "other", "status": "RUNNING"}] + + assert get_backend_job_metadata("http://fl-api-endpoint", "job123") is None + + +@patch("flip_api.fl_services.services.fl_service.http_get") +def test_get_backend_job_metadata_accepts_unknown(mock_http_get): + # UNKNOWN is what an adapter reports for a native status its map does not recognise + # (i.e. a framework upgrade added one). The hub must parse it — and then not act on it + # (the reconcile only acts on FAILED; see test_reconcile_failed_jobs). + from flip_api.fl_services.services.fl_service import get_backend_job_metadata + + mock_http_get.return_value = [{"job_id": "job123", "status": "UNKNOWN"}] + + assert get_backend_job_metadata("http://fl-api-endpoint", "job123").status == FLJobStatus.UNKNOWN + + +@patch("flip_api.fl_services.services.fl_service.http_get") +def test_get_backend_job_metadata_carries_status_details(mock_http_get): + # The backend's own one-line cause rides the same response the status does — that is why + # the lookup returns the whole contract item rather than just the status (#1001). + from flip_api.fl_services.services.fl_service import get_backend_job_metadata + + mock_http_get.return_value = [ + { + "job_id": "job123", + "status": "FAILED", + "status_details": "ServerApp failed with exception: boom", + } + ] + + job = get_backend_job_metadata("http://fl-api-endpoint", "job123") + assert job.status_details == "ServerApp failed with exception: boom" + + +@patch("flip_api.fl_services.services.fl_service.http_get") +def test_get_backend_job_metadata_tolerates_absent_status_details(mock_http_get): + # An FL API predating the field (or NVFLARE, which has no native equivalent) must still + # validate — the field is optional precisely so deploy order cannot break the sweep. + from flip_api.fl_services.services.fl_service import get_backend_job_metadata + + mock_http_get.return_value = [{"job_id": "job123", "status": "FAILED"}] + + assert get_backend_job_metadata("http://fl-api-endpoint", "job123").status_details is None + + +@patch("flip_api.fl_services.services.fl_service.http_get") +def test_get_backend_job_metadata_rejects_non_list_response(mock_http_get): + from flip_api.fl_services.services.fl_service import get_backend_job_metadata + + mock_http_get.return_value = {"job_id": "job123"} + + with pytest.raises(ValueError, match="Unexpected response format"): + get_backend_job_metadata("http://fl-api-endpoint", "job123") + + +@patch("flip_api.fl_services.services.fl_service.http_get") +def test_fetch_run_logs_returns_the_log_tail(mock_http_get): + from flip_api.fl_services.services.fl_service import fetch_run_logs + + mock_http_get.return_value = {"run_id": "42", "log": "ERROR: Exit Code: 607", "truncated": True} + + assert fetch_run_logs("http://fl-api-endpoint", "42") == "ERROR: Exit Code: 607" + mock_http_get.assert_called_once_with("http://fl-api-endpoint/run_logs/42", timeout=60) + + +@pytest.mark.parametrize( + ("side_effect", "return_value"), + [ + (RuntimeError("no such endpoint"), None), + (None, {"run_id": "42"}), + (None, "not a dict"), + ], +) +@patch("flip_api.fl_services.services.fl_service.http_get") +def test_fetch_run_logs_degrades_to_none(mock_http_get, side_effect, return_value): + # Logs are diagnostic detail, never control flow: an FL API without /run_logs (the + # NVFLARE adapter today), one that is down, or one returning nonsense must all leave + # the caller free to report the failure without them. + from flip_api.fl_services.services.fl_service import fetch_run_logs + + mock_http_get.side_effect = side_effect + mock_http_get.return_value = return_value + + assert fetch_run_logs("http://fl-api-endpoint", "42") is None + + @patch("flip_api.fl_services.services.fl_service.extract_current_job_data") @patch("flip_api.fl_services.services.fl_service.get_fl_backend_job_id_by_model_id") @patch("flip_api.fl_services.services.fl_service.fetch_server_status") diff --git a/flip-api/tests/unit/fl_services/services/test_job_metadata_contract.py b/flip-api/tests/unit/fl_services/services/test_job_metadata_contract.py index 65b37b9ba..3646d5047 100644 --- a/flip-api/tests/unit/fl_services/services/test_job_metadata_contract.py +++ b/flip-api/tests/unit/fl_services/services/test_job_metadata_contract.py @@ -17,15 +17,33 @@ from flip_api.domain.schemas.status import FLJobStatus -def test_fl_job_status_has_exactly_five_contract_values(): - assert {s.value for s in FLJobStatus} == {"PENDING", "RUNNING", "FINISHED", "FAILED", "STOPPED"} +def test_fl_job_status_has_exactly_six_contract_values(): + assert {s.value for s in FLJobStatus} == {"PENDING", "RUNNING", "FINISHED", "FAILED", "STOPPED", "UNKNOWN"} -def test_job_metadata_has_exactly_job_id_and_status(): - assert set(IJobMetaData.model_fields) == {"job_id", "status"} +def test_job_metadata_has_exactly_the_contract_fields(): + # Pinned deliberately: the contract is implemented independently by two FL API adapters, + # so a field added on one side and not the other is invisible until it matters. Adding a + # field here means adding it to both adapters' JobMetadata in the same PR. + assert set(IJobMetaData.model_fields) == {"job_id", "status", "status_details"} -@pytest.mark.parametrize("job_status", ["PENDING", "RUNNING", "FINISHED", "FAILED", "STOPPED"]) +def test_status_details_is_optional_and_defaults_to_none(): + # Optional so an FL API image predating the field still validates against a newer hub — + # unlike the UNKNOWN status value, this half of the contract is deploy-order-safe in both + # directions (the hub also ignores extra fields, so a newer FL API is safe too). + job = IJobMetaData.model_validate({"job_id": "abc", "status": "FAILED"}) + assert job.status_details is None + + +def test_status_details_is_carried_through_when_present(): + job = IJobMetaData.model_validate( + {"job_id": "abc", "status": "FAILED", "status_details": "ServerApp failed with exception: boom"} + ) + assert job.status_details == "ServerApp failed with exception: boom" + + +@pytest.mark.parametrize("job_status", ["PENDING", "RUNNING", "FINISHED", "FAILED", "STOPPED", "UNKNOWN"]) def test_job_metadata_accepts_every_contract_status(job_status): job = IJobMetaData.model_validate({"job_id": "abc", "status": job_status}) assert job.status == FLJobStatus(job_status) diff --git a/flip-api/tests/unit/fl_services/test_reconcile_failed_jobs.py b/flip-api/tests/unit/fl_services/test_reconcile_failed_jobs.py new file mode 100644 index 000000000..b368b5864 --- /dev/null +++ b/flip-api/tests/unit/fl_services/test_reconcile_failed_jobs.py @@ -0,0 +1,274 @@ +# Copyright (c) Guy's and St Thomas' NHS Foundation Trust & King's College London +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# http://www.apache.org/licenses/LICENSE-2.0 +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +from datetime import datetime, timedelta +from unittest.mock import MagicMock, patch +from uuid import uuid4 + +import pytest + +from flip_api.domain.interfaces.fl import IJobMetaData +from flip_api.domain.schemas.status import FLJobStatus, ModelStatus +from flip_api.domain.schemas.types import FLBackend +from flip_api.fl_services.reconcile_failed_jobs import ( + reconcile_failed_fl_jobs, + reconcile_failed_fl_jobs_scheduled_task, +) + +_ENDPOINT = "http://flip-fl-api-net-1:8000" +_BACKEND_JOB_ID = "11536428743664681318" +_GRACE_MINUTES = 30 + + +def _job_row(backend_job_id=_BACKEND_JOB_ID, started=None, endpoint=_ENDPOINT, fl_backend=FLBackend.FLOWER): + """One row of the in-flight-jobs query: (job id, model id, backend job id, started, endpoint, backend). + + ``started`` defaults to just-now — inside the unlisted-run grace period. + """ + if started is None: + started = datetime.utcnow() + return (uuid4(), uuid4(), backend_job_id, started, endpoint, fl_backend) + + +def _mock_db(job_rows, model_statuses=None): + """A session whose first exec() yields the job rows and whose later execs yield model statuses. + + ``model_statuses`` are consumed one per job that reaches the post-status re-read, in order. + """ + jobs_result = MagicMock() + jobs_result.all.return_value = job_rows + + results = [jobs_result] + for model_status in model_statuses or []: + status_result = MagicMock() + status_result.one_or_none.return_value = model_status + results.append(status_result) + + db = MagicMock() + db.exec.side_effect = results + return db + + +def _meta(status, details=None): + """One `GET /list_jobs` item as the FL API would return it.""" + return IJobMetaData(job_id=_BACKEND_JOB_ID, status=status, status_details=details) + + +@pytest.fixture +def mock_dependencies(): + settings = MagicMock() + settings.FL_JOB_UNLISTED_GRACE_MINUTES = _GRACE_MINUTES + with ( + patch("flip_api.fl_services.reconcile_failed_jobs.get_backend_job_metadata") as mock_status, + patch("flip_api.fl_services.reconcile_failed_jobs.fetch_run_logs") as mock_logs, + patch("flip_api.fl_services.reconcile_failed_jobs.add_log") as mock_add_log, + patch("flip_api.fl_services.reconcile_failed_jobs.update_model_status") as mock_update, + patch("flip_api.fl_services.reconcile_failed_jobs.get_settings", return_value=settings), + ): + mock_logs.return_value = None + yield { + "status": mock_status, + "logs": mock_logs, + "add_log": mock_add_log, + "update_model_status": mock_update, + } + + +def test_failed_run_errors_the_model_and_logs_the_cause(mock_dependencies): + job = _job_row() + db = _mock_db([job], model_statuses=[ModelStatus.RUNNING]) + mock_dependencies["status"].return_value = _meta(FLJobStatus.FAILED) + mock_dependencies["logs"].return_value = "ImportError: cannot import name 'min_clients_from_run_config'" + + reported = reconcile_failed_fl_jobs(db) + + assert reported == 1 + mock_dependencies["add_log"].assert_called_once() + log_args, log_kwargs = mock_dependencies["add_log"].call_args + assert log_args[0] == job[1] + assert _BACKEND_JOB_ID in log_args[1] + assert "min_clients_from_run_config" in log_args[1] + assert log_kwargs["success"] is False + # transaction=session defers add_log's commit to update_model_status's own, so a failed + # status write can never leave an orphaned feed row for the next tick to duplicate. + assert log_kwargs["transaction"] is db + mock_dependencies["update_model_status"].assert_called_once_with(job[1], ModelStatus.ERROR, db) + + +@pytest.mark.parametrize( + ("fl_backend", "expected_hint"), + # NVFLARE deliberately no longer names show_errors: that route 500s (FLIP#1032) and the + # job workspace holds no readable log, so the container output is the only real answer. + [(FLBackend.FLOWER, "flwr log"), (FLBackend.NVFLARE, "docker logs fl-server-net-")], +) +def test_failure_without_retrievable_logs_names_the_backend_manual_fallback( + mock_dependencies, fl_backend, expected_hint +): + db = _mock_db([_job_row(fl_backend=fl_backend)], model_statuses=[ModelStatus.INITIATED]) + mock_dependencies["status"].return_value = _meta(FLJobStatus.FAILED) + mock_dependencies["logs"].return_value = None + + reconcile_failed_fl_jobs(db) + + message = mock_dependencies["add_log"].call_args[0][1] + assert expected_hint in message + + +def test_backend_status_details_lead_the_message(mock_dependencies): + """The backend's one-line cause goes above the log tail, not after it. + + A Flower run log opens with the per-run `uv sync` -- one `Installed: [...]` line is most + of the stored row -- so a reader who only sees the top of the feed panel would otherwise + get a dependency manifest instead of the reason. + """ + db = _mock_db([_job_row()], model_statuses=[ModelStatus.RUNNING]) + mock_dependencies["status"].return_value = _meta( + FLJobStatus.FAILED, + details="ServerApp failed with exception: No module named 'flwr.common.message'", + ) + mock_dependencies["logs"].return_value = "Installed: [a==1, b==2]\nTraceback...\nModuleNotFoundError" + + reconcile_failed_fl_jobs(db) + + message = mock_dependencies["add_log"].call_args[0][1] + assert "Reported cause: ServerApp failed with exception: No module named" in message + # Cause first, log tail after -- the ordering is the point. + assert message.index("Reported cause:") < message.index("End of the FL run log:") + + +def test_status_details_are_reported_even_without_a_log(mock_dependencies): + """An NVFLARE-style failure (no /run_logs) still gets any cause the backend supplied.""" + db = _mock_db([_job_row(fl_backend=FLBackend.NVFLARE)], model_statuses=[ModelStatus.INITIATED]) + mock_dependencies["status"].return_value = _meta(FLJobStatus.FAILED, details="job failed to deploy") + mock_dependencies["logs"].return_value = None + + reconcile_failed_fl_jobs(db) + + message = mock_dependencies["add_log"].call_args[0][1] + assert "Reported cause: job failed to deploy" in message + assert "docker logs fl-server-net-" in message + + +def test_absent_status_details_change_nothing(mock_dependencies): + """A backend with no per-job explanation (NVFLARE today) reads exactly as before.""" + db = _mock_db([_job_row()], model_statuses=[ModelStatus.RUNNING]) + mock_dependencies["status"].return_value = _meta(FLJobStatus.FAILED, details=None) + mock_dependencies["logs"].return_value = "Traceback...\nImportError" + + reconcile_failed_fl_jobs(db) + + message = mock_dependencies["add_log"].call_args[0][1] + assert "Reported cause:" not in message + assert "End of the FL run log:" in message + + +@pytest.mark.parametrize( + "backend_status", + [FLJobStatus.RUNNING, FLJobStatus.PENDING, FLJobStatus.FINISHED, FLJobStatus.STOPPED, FLJobStatus.UNKNOWN, None], +) +def test_non_failed_runs_are_left_alone(mock_dependencies, backend_status): + # FINISHED in particular: a run whose ServerApp has finished is routinely still + # uploading results, and the hub's own RESULTS_UPLOADED callback owns that transition. + # UNKNOWN: an unmapped native status must never be acted on. None: within the grace + # period (the row's `started` defaults to just-now) an unlisted run is a listing + # hiccup, not a death. + db = _mock_db([_job_row()]) + mock_dependencies["status"].return_value = _meta(backend_status) if backend_status else None + + reported = reconcile_failed_fl_jobs(db) + + assert reported == 0 + mock_dependencies["add_log"].assert_not_called() + mock_dependencies["update_model_status"].assert_not_called() + mock_dependencies["logs"].assert_not_called() + + +def test_unlisted_run_past_grace_is_resolved_without_a_log_fetch(mock_dependencies): + # A SuperLink restart loses its in-memory run state: the run is gone, its log with it, + # and it can never report. Past the grace period that is a death, not a hiccup. + job = _job_row(started=datetime.utcnow() - timedelta(minutes=_GRACE_MINUTES + 5)) + db = _mock_db([job], model_statuses=[ModelStatus.INITIATED]) + mock_dependencies["status"].return_value = None + + reported = reconcile_failed_fl_jobs(db) + + assert reported == 1 + mock_dependencies["logs"].assert_not_called() + message = mock_dependencies["add_log"].call_args[0][1] + assert "no longer listed" in message + assert _BACKEND_JOB_ID in message + mock_dependencies["update_model_status"].assert_called_once_with(job[1], ModelStatus.ERROR, db) + + +def test_unlisted_run_with_no_started_timestamp_is_left_alone(mock_dependencies): + # `started` is set at job pickup so an in-flight job should always carry one; if it + # somehow doesn't, there is nothing to measure the grace period from — do nothing. + job_id, model_id, backend_job_id, _, endpoint, fl_backend = _job_row() + db = _mock_db([(job_id, model_id, backend_job_id, None, endpoint, fl_backend)]) + mock_dependencies["status"].return_value = None + + reported = reconcile_failed_fl_jobs(db) + + assert reported == 0 + mock_dependencies["add_log"].assert_not_called() + mock_dependencies["update_model_status"].assert_not_called() + + +def test_no_in_flight_jobs_does_not_call_the_fl_api(mock_dependencies): + db = _mock_db([]) + + assert reconcile_failed_fl_jobs(db) == 0 + mock_dependencies["status"].assert_not_called() + + +@pytest.mark.parametrize( + "settled_status", + [ModelStatus.RESULTS_UPLOADED, ModelStatus.STOPPED, ModelStatus.ERROR], +) +def test_model_that_settled_during_the_status_call_wins_the_race(mock_dependencies, settled_status): + db = _mock_db([_job_row()], model_statuses=[settled_status]) + mock_dependencies["status"].return_value = _meta(FLJobStatus.FAILED) + + reported = reconcile_failed_fl_jobs(db) + + assert reported == 0 + mock_dependencies["add_log"].assert_not_called() + mock_dependencies["update_model_status"].assert_not_called() + + +def test_unlisted_run_past_grace_still_defers_to_a_settled_model(mock_dependencies): + job = _job_row(started=datetime.utcnow() - timedelta(minutes=_GRACE_MINUTES + 5)) + db = _mock_db([job], model_statuses=[ModelStatus.STOPPED]) + mock_dependencies["status"].return_value = None + + reported = reconcile_failed_fl_jobs(db) + + assert reported == 0 + mock_dependencies["add_log"].assert_not_called() + mock_dependencies["update_model_status"].assert_not_called() + + +def test_one_unreachable_net_does_not_stop_the_other_jobs(mock_dependencies): + unreachable, healthy = _job_row(), _job_row(backend_job_id="222") + db = _mock_db([unreachable, healthy], model_statuses=[ModelStatus.RUNNING]) + mock_dependencies["status"].side_effect = [ConnectionError("net-1 is down"), _meta(FLJobStatus.FAILED)] + + reported = reconcile_failed_fl_jobs(db) + + assert reported == 1 + db.rollback.assert_called_once() + mock_dependencies["update_model_status"].assert_called_once_with(healthy[1], ModelStatus.ERROR, db) + + +def test_scheduled_task_never_raises(): + with patch("flip_api.fl_services.reconcile_failed_jobs.get_engine", side_effect=RuntimeError("no db")): + reconcile_failed_fl_jobs_scheduled_task() diff --git a/flip-api/tests/unit/test_config.py b/flip-api/tests/unit/test_config.py index 322adf547..b0409a792 100644 --- a/flip-api/tests/unit/test_config.py +++ b/flip-api/tests/unit/test_config.py @@ -78,6 +78,9 @@ def test_scan_int_settings_empty_string_falls_back_to_default(): assert Settings(PICKLESCAN_TIMEOUT_SECONDS="45").PICKLESCAN_TIMEOUT_SECONDS == 45 assert Settings(BANDIT_TIMEOUT_SECONDS="").BANDIT_TIMEOUT_SECONDS == 60 assert Settings(BANDIT_TIMEOUT_SECONDS="30").BANDIT_TIMEOUT_SECONDS == 30 + assert Settings(SCHEDULER_FL_JOB_RECONCILE_RATE="").SCHEDULER_FL_JOB_RECONCILE_RATE == 1 + assert Settings(FL_JOB_UNLISTED_GRACE_MINUTES="").FL_JOB_UNLISTED_GRACE_MINUTES == 30 + assert Settings(FL_JOB_UNLISTED_GRACE_MINUTES="45").FL_JOB_UNLISTED_GRACE_MINUTES == 45 def test_suffix_list_passes_through_unexpected_types_for_pydantic_to_reject(): diff --git a/flip-ui/src/partials/models/Timeline.vue b/flip-ui/src/partials/models/Timeline.vue index a3e23c374..3c0af5530 100644 --- a/flip-ui/src/partials/models/Timeline.vue +++ b/flip-ui/src/partials/models/Timeline.vue @@ -61,7 +61,14 @@ {{ getShortDateFromString(log.logDate) }} -

+ +

{{ log.log }}

diff --git a/flip-ui/src/partials/models/__tests__/Timeline.spec.ts b/flip-ui/src/partials/models/__tests__/Timeline.spec.ts index 0720831e9..b97ee50e9 100644 --- a/flip-ui/src/partials/models/__tests__/Timeline.spec.ts +++ b/flip-ui/src/partials/models/__tests__/Timeline.spec.ts @@ -118,4 +118,22 @@ describe("Timeline", () => { expect(stamp.classes()).toContain("dark:text-gray-300"); } }); + + test("preserves the line structure of failure logs, and only failure logs", () => { + // A failed-run row carries the FL run's log tail (FLIP#1001) — a multi-line + // traceback whose line breaks are the readability. Without pre-wrap the browser + // collapses it to one wall of wrapped text. Success rows are single-line prose + // and keep the default so stray whitespace never reformats them. + const comp = mount(Timeline, { + props: { complete: true }, + global: { stubs: { AiLoader: true } } + }); + + const texts = comp.findAll("[data-test='log-text']"); + expect(texts).toHaveLength(mockLogs.logs.length); + const failureRow = texts.find((t) => t.text().includes("Round failed")); + const successRow = texts.find((t) => t.text().includes("Training started")); + expect(failureRow?.classes()).toContain("whitespace-pre-wrap"); + expect(successRow?.classes()).not.toContain("whitespace-pre-wrap"); + }); });