Wave 1 post-merge audit closure (2 commits)

docs/roadmap.md

@@ -4,7 +4,7 @@
 
 ## Status at a glance
 
-| Tür | Phase | Status |
+| Type | Phase | Status |
 |-----|-------|--------|
 | ✅ Done | [Phase 1-9](roadmap/completed-phases.md) | SOTA upgrades, evaluation, reliability, enterprise integration, ecosystem, alignment stack, safety, EU AI Act compliance (Articles 9-17 + Annex IV), advanced safety intelligence |
 | ✅ Done | [Phase 10 — Post-Training Completion](roadmap/phase-10-post-training.md) | `inference.py`, `chat`, `export` (GGUF), `--fit-check`, `deploy` — shipped `v0.4.0` |

forgelm/benchmark.py

@@ -93,7 +93,12 @@ def _save_benchmark_json(
                 indent=2,
             )
         logger.info("Benchmark results saved to %s", results_path)
-    except Exception as e:
+    except (OSError, TypeError, ValueError) as e:
+        # OSError: filesystem (ENOSPC, permission, broken parent dir).
+        # TypeError/ValueError: ``json.dump`` rejecting an unserialisable
+        # object inside the lm-eval result tree.  Saving the artefact is
+        # non-fatal: the run already completed and metrics live in the
+        # returned BenchmarkResult.
         logger.warning("Failed to save benchmark results: %s", e)
 
 
@@ -135,7 +140,7 @@ def run_benchmark(
 
     try:
         lm_obj = HFLM(pretrained=model, tokenizer=tokenizer, batch_size=batch_size)
-    except Exception as e:
+    except Exception as e:  # noqa: BLE001 — best-effort: lm-eval HFLM wrapper construction crosses HF model introspection (AttributeError on architecture mismatch), tokenizer compatibility (ValueError), CUDA init (RuntimeError), and lm-eval-internal config parsing; surfacing as BenchmarkResult(passed=False) is the documented hard-failure surface so the trainer auto-revert gate can react.  # NOSONAR
         logger.error("Failed to initialize lm-eval model wrapper: %s", e)
         return BenchmarkResult(passed=False, failure_reason=f"Model wrapper initialization failed: {e}")
 
@@ -145,7 +150,7 @@ def run_benchmark(
 
     try:
         results = lm_eval.simple_evaluate(model=lm_obj, tasks=tasks, limit=limit, **task_kwargs)
-    except Exception as e:
+    except Exception as e:  # noqa: BLE001 — best-effort: lm-eval simple_evaluate runs a wide task surface (dataset download OSError, task spec ValueError, model.generate RuntimeError on CUDA OOM/dtype mismatch, lm-eval-internal AssertionError); BenchmarkResult(passed=False) is the documented hard-failure surface for the auto-revert gate.  # NOSONAR
         logger.error("Benchmark evaluation failed: %s", e)
         return BenchmarkResult(passed=False, failure_reason=f"Evaluation execution failed: {e}")
 

forgelm/chat.py

@@ -271,7 +271,7 @@ def _generate_and_print(self, user_input: str) -> None:
                 ):
                     self._print_inline_raw(token)
                     response += token
-            except Exception as e:
+            except Exception as e:  # noqa: BLE001 — best-effort: streaming generate crosses CUDA OOM (RuntimeError), tokenizer issues (ValueError), and stop-token race conditions; the chat REPL must keep running after a single failed turn so the operator can retry / change parameters.  # NOSONAR
                 self._print(f"\n[Generation error: {e}]")
                 return
             self._print("")  # trailing newline
@@ -285,7 +285,7 @@ def _generate_and_print(self, user_input: str) -> None:
                     temperature=self.temperature,
                     max_new_tokens=self.max_new_tokens,
                 )
-            except Exception as e:
+            except Exception as e:  # noqa: BLE001 — best-effort: see streaming branch above; same broad torch / tokenizer surface, same per-turn soft-fail policy in the REPL loop.  # NOSONAR
                 self._print(f"[Generation error: {e}]")
                 return
             # Non-streaming path: full response is also untrusted model output

forgelm/cli/_training.py

@@ -70,7 +70,7 @@ def _run_training_pipeline(config, args, json_output: bool) -> None:
             sys.exit(EXIT_AWAITING_APPROVAL)
         sys.exit(EXIT_SUCCESS if result.success else EXIT_EVAL_FAILURE)
 
-    except Exception as e:
+    except Exception as e:  # noqa: BLE001 — best-effort: top-of-CLI training catch.  ForgeTrainer.train() crosses every concern (HF model load, dataset load, TRL trainer, safety eval, judge eval, audit emission, compliance export, webhook notify); any leak from the inner narrow-class catches must surface as a structured CLI failure (with traceback in the log) rather than a Python traceback dumped to stdout that breaks JSON parsers.  # NOSONAR
         _report_training_error(
             json_output,
             payload={"success": False, "error": str(e)},

forgelm/cli/subcommands/_chat.py

@@ -29,6 +29,6 @@ def _run_chat_cmd(args) -> None:
             trust_remote_code=args.trust_remote_code,
             backend=args.backend,
         )
-    except Exception as e:
+    except Exception as e:  # noqa: BLE001 — best-effort: top-of-subcommand catch.  ``run_chat`` loads the model + tokenizer, optionally a PEFT adapter, and runs the REPL loop; failure surface includes OSError (model path), RuntimeError (CUDA), ValueError (tokenizer), KeyboardInterrupt swap, plus chat-internal generation errors.  EXIT_TRAINING_ERROR is the documented public contract for a failed dispatched subcommand.  # NOSONAR
         logger.exception("Chat session failed: %s", e)
         sys.exit(EXIT_TRAINING_ERROR)

forgelm/config.py

@@ -310,14 +310,32 @@
     )
 
 
+# EU AI Act risk taxonomy — single source of truth shared by
+# ``RiskAssessmentConfig.risk_category`` and
+# ``ComplianceMetadataConfig.risk_classification`` so the two Pydantic fields
+# can never drift.  ``unacceptable`` covers Article 5 prohibited practices;
+# ``high-risk`` covers Article 6 systems requiring full Annex IV documentation;
+# ``limited-risk`` and ``minimal-risk`` cover the transparency-only and
+# unrestricted tiers; ``unknown`` is the explicit placeholder for systems that
+# have not yet been classified.  The default for both fields stays
+# ``"minimal-risk"`` so existing configs validate unchanged.
+RiskTier = Literal["unknown", "minimal-risk", "limited-risk", "high-risk", "unacceptable"]
+
+# Tiers that demand full Annex IV documentation + auto-revert + safety gates
+# under the EU AI Act.  Keep this set in lockstep with
+# ``ForgeConfig._warn_high_risk_compliance`` and the wizard prompt so the new
+# tier is reachable + enforced everywhere the old high-risk-only set was.
+_STRICT_RISK_TIERS: frozenset[str] = frozenset({"high-risk", "unacceptable"})
+
+
 class RiskAssessmentConfig(BaseModel):
     """Art. 9: Risk management — declare risks before training."""
 
     model_config = ConfigDict(extra="forbid")
 
     intended_use: str = ""
     foreseeable_misuse: List[str] = []
-    risk_category: Literal["high-risk", "limited-risk", "minimal-risk"] = "minimal-risk"
+    risk_category: RiskTier = "minimal-risk"
     mitigation_measures: List[str] = []
     vulnerable_groups_considered: bool = False
 
@@ -346,7 +364,7 @@
     intended_purpose: str = ""
     known_limitations: str = ""
     system_version: str = ""
-    risk_classification: Literal["high-risk", "limited-risk", "minimal-risk"] = "minimal-risk"
+    risk_classification: RiskTier = "minimal-risk"
 
 
 class SyntheticConfig(BaseModel):
@@ -477,17 +495,39 @@
                 self.training.save_steps,
             )
 
     def _warn_high_risk_compliance(self) -> None:
-        """EU AI Act high-risk compliance recommendations."""
-        is_high_risk = (self.risk_assessment and self.risk_assessment.risk_category == "high-risk") or (
-            self.compliance and self.compliance.risk_classification == "high-risk"
+        """EU AI Act compliance recommendations for strict risk tiers.
+
+        ``unacceptable`` (Article 5 prohibited practices) is treated at least
+        as strictly as ``high-risk`` for ForgeLM's purposes — the gate exists
+        to nudge the operator into running with auto-revert + safety eval,
+        and ``unacceptable`` should never get *less* gating than ``high-risk``
+        because the underlying use case is not allowed at all under the Act.
+        """
+        is_strict = (self.risk_assessment and self.risk_assessment.risk_category in _STRICT_RISK_TIERS) or (
+            self.compliance and self.compliance.risk_classification in _STRICT_RISK_TIERS
         )
-        if not is_high_risk:
+        if not is_strict:
             return
         if not self.evaluation or not self.evaluation.auto_revert:
             logger.warning(
-                "High-risk AI classification requires evaluation.auto_revert: true "
-                "for EU AI Act compliance. Safety gates should be enabled."
+                "Risk classification %r requires evaluation.auto_revert: true "
+                "for EU AI Act compliance. Safety gates should be enabled.",
+                # Use whichever field is set; both share the same RiskTier.
+                (self.compliance and self.compliance.risk_classification)
+                or (self.risk_assessment and self.risk_assessment.risk_category),
+            )
+        # Article 5 prohibited practices warrant a louder operator notice on
+        # top of the auto_revert nudge — the deployment itself is unlawful in
+        # the EU regardless of how well the safety gates are wired up.
+        unacceptable = (self.risk_assessment and self.risk_assessment.risk_category == "unacceptable") or (
+            self.compliance and self.compliance.risk_classification == "unacceptable"
+        )
+        if unacceptable:
+            logger.warning(
+                "Risk classification 'unacceptable' corresponds to EU AI Act Article 5 prohibited "
+                "practices. ForgeLM will not refuse the run, but deploying such a system inside the "
+                "EU is unlawful — confirm operator intent before continuing."
             )
         safety = self.evaluation.safety if self.evaluation else None
         if not safety or not safety.enabled:

forgelm/deploy.py

@@ -297,6 +297,6 @@ def generate_deploy_config(
         logger.info("Deploy config written to %s", output_path)
         return DeployResult(success=True, target=target, output_path=output_path, content=content)
 
-    except Exception as e:
+    except Exception as e:  # noqa: BLE001 — best-effort: deploy-config generators cross filesystem writes (OSError), template rendering (KeyError/ValueError), and target-specific YAML/JSON serialisation; surfacing every variant as a structured DeployResult is the documented public contract.  # NOSONAR
         logger.error("Failed to generate deploy config for target '%s': %s", target, e)
         return DeployResult(success=False, target=target, error=str(e))

forgelm/export.py

@@ -148,8 +148,14 @@ def _merge_adapter(model_path: str, adapter_path: str, merged_dir: str) -> None:
     from transformers import AutoModelForCausalLM, AutoTokenizer
 
     logger.info("Merging adapter %s into base model %s ...", adapter_path, model_path)
-    tokenizer = AutoTokenizer.from_pretrained(model_path)
-    model = AutoModelForCausalLM.from_pretrained(model_path, torch_dtype=torch.float16, device_map="cpu")
+    # ``trust_remote_code=False`` is the secure default (Phase 7 acceptance):
+    # the merge step is part of the GGUF export pipeline; loading the base
+    # model must not execute repo-bundled code.  Operators with a custom
+    # architecture should fork and pre-convert before exporting.
+    tokenizer = AutoTokenizer.from_pretrained(model_path, trust_remote_code=False)
+    model = AutoModelForCausalLM.from_pretrained(
+        model_path, torch_dtype=torch.float16, device_map="cpu", trust_remote_code=False
+    )
     model = PeftModel.from_pretrained(model, adapter_path)
     model = model.merge_and_unload()
 
@@ -203,7 +209,13 @@ def _update_integrity_manifest(model_dir: str, export_result: ExportResult) -> N
             json.dump(data, f, indent=2)
 
         logger.info("model_integrity.json updated with exported artifact.")
-    except Exception as e:
+    except (OSError, TypeError, ValueError) as e:
+        # OSError: filesystem write / read failure on integrity_path.
+        # TypeError: ``json.dump`` rejecting an unserialisable artefact field.
+        # ValueError: corrupt or partial existing manifest (covers
+        # ``json.JSONDecodeError`` since it subclasses ``ValueError``).
+        # Updating the manifest is non-fatal: the artefact itself was already
+        # exported successfully and its on-disk SHA-256 is recoverable.
         logger.debug("Could not update model_integrity.json: %s", e)
 
 
@@ -299,7 +311,7 @@ def _run_converter(cmd: List[str], fmt: str, actual_quant: str) -> Optional[Expo
             quant=actual_quant,
             error="GGUF conversion timed out after 3600 seconds.",
         )
-    except Exception as e:
+    except Exception as e:  # noqa: BLE001 — best-effort: subprocess.run for the GGUF converter (llama.cpp / convert-hf-to-gguf.py) crosses OSError (binary missing), CalledProcessError (non-zero exit before check=False catches it), and the converter's own deep-stack errors; ExportResult(success=False) is the documented public contract for the export pipeline.  # NOSONAR
         return ExportResult(success=False, format=fmt, quant=actual_quant, error=str(e))
 
     if proc.returncode == 0:
@@ -385,7 +397,7 @@ def export_model(
         try:
             _merge_adapter(model_path, adapter, merged_dir)
             source_path = merged_dir
-        except Exception as e:
+        except Exception as e:  # noqa: BLE001 — best-effort: _merge_adapter loads HF base + PEFT adapter and writes the merged dir; failure surface includes OSError (disk/path), RuntimeError (CUDA/dtype), KeyError (missing adapter config), and HF/PEFT-internal errors.  ExportResult(success=False) is the documented hard-failure contract.  # NOSONAR
             return ExportResult(success=False, format=fmt, quant=quant, error=f"Adapter merge failed: {e}")
 
     actual_quant, actual_output_path = _resolve_kquant_path(quant, output_path)

forgelm/fit_check.py

@@ -88,7 +88,7 @@ def _load_arch_params(model_name_or_path: str, trust_remote_code: bool = False)
         params["num_attention_heads"] = getattr(cfg, "num_attention_heads", None)
         params["num_key_value_heads"] = getattr(cfg, "num_key_value_heads", None)
         logger.debug("Loaded architecture config from %s", model_name_or_path)
-    except Exception as e:
+    except Exception as e:  # noqa: BLE001 — best-effort: AutoConfig.from_pretrained surfaces OSError (network/cache miss), HF repo errors, ValueError on unknown architecture, and ImportError when the optional config-class module is missing.  fit-check fallback uses regex name-hints (3b/7b/13b/...) so a config probe failure does not abort the VRAM estimate.  # NOSONAR
         logger.debug("Could not load AutoConfig for %s: %s — using size hint fallback.", model_name_or_path, e)
 
     # Fill in missing values using size-hint lookup on the model name.
@@ -299,7 +299,7 @@ def _detect_available_vram_gb(torch_module: Any) -> tuple[Optional[float], bool]
             free_bytes, _ = torch_module.cuda.mem_get_info()
             return free_bytes / (1024**3), False
         return None, True
-    except Exception as e:
+    except Exception as e:  # noqa: BLE001 — best-effort: ``torch.cuda.mem_get_info`` surfaces RuntimeError (driver / CUDA init), AttributeError (older torch / no CUDA), and OSError (corrupt driver state).  Returning ``(None, True)`` falls back to a hypothetical-only fit verdict.  # NOSONAR
         logger.debug("Could not query GPU memory: %s", e)
         return None, True
 

forgelm/inference.py

@@ -289,7 +289,7 @@ def _gen_thread() -> None:
         try:
             with torch.inference_mode():
                 model.generate(**gen_kwargs)
-        except Exception as e:
+        except Exception as e:  # noqa: BLE001 — best-effort: streaming generate runs in a worker thread; the broad catch funnels every torch error (CUDA OOM RuntimeError, dtype mismatch, tokenizer ValueError, KeyboardInterrupt swap) into the shared list so the foreground thread can re-raise after joining.  # NOSONAR
             _exc.append(e)
 
     thread = Thread(target=_gen_thread, daemon=True)

forgelm/ingestion.py

@@ -227,7 +227,7 @@ def _read_pdf_pages(reader: Any, path: Path) -> List[str]:
     for idx, page in enumerate(reader.pages):
         try:
             text = page.extract_text() or ""
-        except Exception as exc:
+        except Exception as exc:  # noqa: BLE001 — best-effort: pypdf's per-page extraction surface is wide (KeyError on malformed object refs, AssertionError on broken cross-ref tables, UnicodeDecodeError on font encodings, plus its own internal errors); per-page soft-fail keeps the run going so a single bad page cannot abort a multi-thousand-document corpus ingest.  # NOSONAR
             logger.warning(
                 "PDF page extraction failed (file=%s, page_index=%d): %s — page skipped, run continues.",
                 path,
@@ -257,7 +257,7 @@ def _extract_pdf(path: Path, *, dedup_state: Optional[Dict[str, int]] = None) ->
 
     try:
         reader = PdfReader(str(path))
-    except Exception as exc:
+    except Exception as exc:  # noqa: BLE001 — best-effort: PdfReader open surfaces OSError (file/permission), pypdf-internal errors (bad header, unsupported version, malformed xref), and on rare adversarial inputs InfiniteLoopError; converting all to a typed ValueError keeps the per-file error path uniform with DOCX/EPUB extractors.  # NOSONAR
         raise ValueError(f"Could not open PDF '{path}': {exc}") from exc
 
     if getattr(reader, "is_encrypted", False):
@@ -1055,7 +1055,7 @@ def _extract_text_for_ingest(
         return extractor(fpath)
     except ImportError:
         raise
-    except Exception as exc:
+    except Exception as exc:  # noqa: BLE001 — best-effort: per-file dispatcher catch — each format-specific extractor (PDF/DOCX/EPUB/TXT/MD) raises its own typed ValueError above + a wide tail of corpus-data-driven failures; per-file soft-fail with skip-and-continue keeps a multi-thousand-file corpus ingest running.  ImportError stays narrow above so missing-extra failures still propagate.  # NOSONAR
         logger.warning("Skipping '%s' (extraction failed): %s", fpath, exc)
         return None
 

forgelm/judge.py

@@ -163,8 +163,12 @@ def _load_local_judge(judge_model: str) -> Tuple[Any, Any]:
     from transformers import AutoModelForCausalLM, AutoTokenizer
 
     logger.info("Loading local judge model: %s", judge_model)
-    tok = AutoTokenizer.from_pretrained(judge_model)
-    mdl = AutoModelForCausalLM.from_pretrained(judge_model, device_map="auto")
+    # ``trust_remote_code=False`` is the secure default (Phase 7 acceptance): a
+    # judge model is consulted by the auto-revert gate, so loading must not
+    # execute arbitrary repo code at load time.  Operators with a custom
+    # architecture should fork and pre-convert.
+    tok = AutoTokenizer.from_pretrained(judge_model, trust_remote_code=False)
+    mdl = AutoModelForCausalLM.from_pretrained(judge_model, device_map="auto", trust_remote_code=False)
     return mdl, tok
 
 

forgelm/merging.py

@@ -79,7 +79,7 @@ def merge_peft_adapters(
             num_models=len(adapters),
         )
 
-    except Exception as e:
+    except Exception as e:  # noqa: BLE001 — best-effort: model merging crosses HF model load (OSError/RuntimeError), PEFT adapter load (KeyError on missing config), torch tensor ops (RuntimeError on dtype/device mismatch), and mergekit-internal errors.  MergeResult(success=False) is the documented public contract.  # NOSONAR
         logger.error("Model merging failed: %s", e)
         return MergeResult(success=False, error=str(e))