diff --git a/skills/cdopt-optimization/scripts/write_constrained_layer_runner.py b/skills/cdopt-optimization/scripts/write_constrained_layer_runner.py index 07e33a3..34358e3 100644 --- a/skills/cdopt-optimization/scripts/write_constrained_layer_runner.py +++ b/skills/cdopt-optimization/scripts/write_constrained_layer_runner.py @@ -2,6 +2,7 @@ """Write a tiny CPU-only CDOpt PyTorch constrained-layer training runner.""" import argparse +import math import textwrap from pathlib import Path @@ -17,6 +18,7 @@ import argparse import json +import math import time from pathlib import Path @@ -61,6 +63,23 @@ def main(): parser.add_argument("--results-dir", default="results") args = parser.parse_args() + positive_ints = { + "--in-features": args.in_features, + "--hidden-features": args.hidden_features, + "--num-classes": args.num_classes, + "--batch": args.batch, + "--steps": args.steps, + } + for name, value in positive_ints.items(): + if value < 1 or value > 100_000: + parser.error(f"{name} must be between 1 and 100000") + for name, value in (("--lr", args.lr), ("--penalty", args.penalty)): + if not math.isfinite(value) or value <= 0.0: + parser.error(f"{name} must be finite and greater than 0") + results_dir = Path(args.results_dir) + if results_dir.exists() and (results_dir.is_symlink() or not results_dir.is_dir()): + parser.error("--results-dir must be a real directory") + torch.manual_seed(args.seed) rng = np.random.default_rng(args.seed) device = torch.device("cpu") @@ -82,16 +101,20 @@ def main(): initial_loss = None final_loss = None error = None + completed_steps = 0 try: for step in range(args.steps): optimizer.zero_grad() logits = model(x) loss = F.nll_loss(logits, y) + get_quad_penalty(model) + if not torch.isfinite(loss): + raise FloatingPointError("training loss is not finite") loss.backward() optimizer.step() if step == 0: initial_loss = float(loss.item()) final_loss = float(loss.item()) + completed_steps = step + 1 except Exception as exc: # noqa: BLE001 - keep run summary robust error = f"{type(exc).__name__}: {exc}" elapsed = time.time() - started @@ -102,15 +125,23 @@ def main(): except Exception as exc: # noqa: BLE001 - keep run summary robust feasibility = f"unavailable: {type(exc).__name__}: {exc}" + execution_success = error is None and completed_steps == args.steps + verification_success = isinstance(feasibility, float) and math.isfinite(feasibility) summary = { "example": "lenet_style_stiefel_constrained_layer_torch", "framework": "pytorch + cdopt.nn constraint-dissolving layer", - "success": error is None, + "success": execution_success and verification_success, + "failure_stage": None if error is None else "training", + "execution": {"success": execution_success, "completed_steps": completed_steps}, + "solver": {"success": execution_success, "kind": "torch.optim.SGD"}, + "verification": {"success": verification_success, "metric": "quadratic_penalty_proxy"}, + "mathematical_conclusion": "not_assessed", "error": error, "initial_loss": initial_loss, "final_loss": final_loss, "final_quad_penalty": feasibility, "steps": args.steps, + "completed_steps": completed_steps, "elapsed_seconds": elapsed, "parameters": { "in_features": args.in_features, @@ -133,16 +164,20 @@ def main(): }, } - results_dir = Path(args.results_dir) results_dir.mkdir(parents=True, exist_ok=True) + if results_dir.is_symlink(): + raise RuntimeError("results directory became a symlink") out_path = results_dir / "solver_summary.json" + if out_path.is_symlink(): + raise RuntimeError("refusing to replace symlinked solver_summary.json") out_path.write_text(json.dumps(summary, indent=2, sort_keys=True)) print(json.dumps(summary, indent=2, sort_keys=True)) print(f"wrote {out_path}") + return 0 if summary["success"] else 1 if __name__ == "__main__": - main() + raise SystemExit(main()) ''' @@ -156,8 +191,12 @@ def main(): args = parser.parse_args() output_dir = Path(args.output_dir) + if output_dir.is_symlink() or (output_dir.exists() and not output_dir.is_dir()): + parser.error("--output-dir must be a real directory") output_dir.mkdir(parents=True, exist_ok=True) runner_path = output_dir / "run_constrained_layer.py" + if runner_path.is_symlink() or (runner_path.exists() and not runner_path.is_file()): + parser.error("refusing to replace an unsafe generated runner path") runner_path.write_text(textwrap.dedent(RUNNER)) runner_path.chmod(0o755) print(runner_path) diff --git a/skills/cdopt-optimization/scripts/write_constrained_rnn_runner.py b/skills/cdopt-optimization/scripts/write_constrained_rnn_runner.py index 475374e..dbf1b5c 100644 --- a/skills/cdopt-optimization/scripts/write_constrained_rnn_runner.py +++ b/skills/cdopt-optimization/scripts/write_constrained_rnn_runner.py @@ -2,6 +2,7 @@ """Write a tiny CPU-only CDOpt PyTorch constrained RNN/LSTM training runner.""" import argparse +import math import textwrap from pathlib import Path @@ -16,6 +17,7 @@ import argparse import json +import math import time from pathlib import Path @@ -123,6 +125,25 @@ def main(): parser.add_argument("--results-dir", default="results") args = parser.parse_args() + positive_ints = { + "--batch": args.batch, + "--seq-len": args.seq_len, + "--input-size": args.input_size, + "--hidden-size": args.hidden_size, + "--num-layers": args.num_layers, + "--num-classes": args.num_classes, + "--steps": args.steps, + } + for name, value in positive_ints.items(): + if value < 1 or value > 100_000: + parser.error(f"{name} must be between 1 and 100000") + for name, value in (("--lr", args.lr), ("--penalty", args.penalty)): + if not math.isfinite(value) or value <= 0.0: + parser.error(f"{name} must be finite and greater than 0") + results_dir = Path(args.results_dir) + if results_dir.exists() and (results_dir.is_symlink() or not results_dir.is_dir()): + parser.error("--results-dir must be a real directory") + torch.manual_seed(args.seed) rng = np.random.default_rng(args.seed) device = torch.device("cpu") @@ -143,16 +164,20 @@ def main(): initial_loss = None final_loss = None error = None + completed_steps = 0 try: for step in range(args.steps): optimizer.zero_grad() logits = model(x) loss = criterion(logits, y) + get_quad_penalty(model) + if not torch.isfinite(loss): + raise FloatingPointError("training loss is not finite") loss.backward() optimizer.step() if step == 0: initial_loss = float(loss.item()) final_loss = float(loss.item()) + completed_steps = step + 1 except Exception as exc: # noqa: BLE001 - keep run summary robust error = f"{type(exc).__name__}: {exc}" elapsed = time.time() - started @@ -163,15 +188,23 @@ def main(): except Exception as exc: # noqa: BLE001 - keep run summary robust feasibility = f"unavailable: {type(exc).__name__}: {exc}" + execution_success = error is None and completed_steps == args.steps + verification_success = isinstance(feasibility, float) and math.isfinite(feasibility) summary = { "example": f"constrained_{args.cell_type}_torch", "framework": f"pytorch + cdopt.nn {args.cell_type.upper()}_cdopt", - "success": error is None, + "success": execution_success and verification_success, + "failure_stage": None if error is None else "training", + "execution": {"success": execution_success, "completed_steps": completed_steps}, + "solver": {"success": execution_success, "kind": "torch.optim.SGD"}, + "verification": {"success": verification_success, "metric": "quadratic_penalty_proxy"}, + "mathematical_conclusion": "not_assessed", "error": error, "initial_loss": initial_loss, "final_loss": final_loss, "final_quad_penalty": feasibility, "steps": args.steps, + "completed_steps": completed_steps, "elapsed_seconds": elapsed, "parameters": { "cell_type": args.cell_type, @@ -198,16 +231,20 @@ def main(): }, } - results_dir = Path(args.results_dir) results_dir.mkdir(parents=True, exist_ok=True) + if results_dir.is_symlink(): + raise RuntimeError("results directory became a symlink") out_path = results_dir / "solver_summary.json" + if out_path.is_symlink(): + raise RuntimeError("refusing to replace symlinked solver_summary.json") out_path.write_text(json.dumps(summary, indent=2, sort_keys=True)) print(json.dumps(summary, indent=2, sort_keys=True)) print(f"wrote {out_path}") + return 0 if summary["success"] else 1 if __name__ == "__main__": - main() + raise SystemExit(main()) ''' @@ -221,8 +258,12 @@ def main(): args = parser.parse_args() output_dir = Path(args.output_dir) + if output_dir.is_symlink() or (output_dir.exists() and not output_dir.is_dir()): + parser.error("--output-dir must be a real directory") output_dir.mkdir(parents=True, exist_ok=True) runner_path = output_dir / "run_constrained_rnn.py" + if runner_path.is_symlink() or (runner_path.exists() and not runner_path.is_file()): + parser.error("refusing to replace an unsafe generated runner path") runner_path.write_text(textwrap.dedent(RUNNER)) runner_path.chmod(0o755) print(runner_path) diff --git a/skills/cdopt-optimization/scripts/write_stiefel_dictionary_runner.py b/skills/cdopt-optimization/scripts/write_stiefel_dictionary_runner.py index 58ad2b7..36f8cf1 100755 --- a/skills/cdopt-optimization/scripts/write_stiefel_dictionary_runner.py +++ b/skills/cdopt-optimization/scripts/write_stiefel_dictionary_runner.py @@ -2,6 +2,7 @@ """Write a tiny CPU-only CDOpt Stiefel dictionary-learning runner.""" import argparse +import math import textwrap from pathlib import Path @@ -11,6 +12,7 @@ import argparse import json +import math import time from pathlib import Path @@ -40,6 +42,22 @@ def main(): parser.add_argument("--results-dir", default="results") args = parser.parse_args() + if args.n < 1 or args.n > 512: + parser.error("--n must be between 1 and 512") + if args.m is not None and not 1 <= args.m <= 10_000_000: + parser.error("--m must be between 1 and 10000000") + if not math.isfinite(args.theta) or not 0.0 <= args.theta <= 1.0: + parser.error("--theta must be finite and between 0 and 1") + if not math.isfinite(args.mu) or args.mu <= 0.0: + parser.error("--mu must be finite and greater than 0") + if args.maxiter < 1 or args.maxiter > 100_000: + parser.error("--maxiter must be between 1 and 100000") + if not math.isfinite(args.gtol) or args.gtol <= 0.0: + parser.error("--gtol must be finite and greater than 0") + results_dir = Path(args.results_dir) + if results_dir.exists() and (results_dir.is_symlink() or not results_dir.is_dir()): + parser.error("--results-dir must be a real directory") + n = args.n m = args.m or 10 * n * n device = torch.device("cpu") @@ -73,10 +91,18 @@ def obj_fun(x): except Exception as exc: # noqa: BLE001 - keep run summary robust feasibility = f"unavailable: {type(exc).__name__}: {exc}" + execution_success = bool(result.success) and all( + math.isfinite(value) for value in (float(result.fun), float(np.linalg.norm(grad))) + ) + verification_success = isinstance(feasibility, float) and math.isfinite(feasibility) summary = { "example": "stiefel_dictionary_learning_torch_scipy", "solver": "scipy.optimize.minimize L-BFGS-B via CDOpt CDF callbacks", - "success": bool(result.success), + "success": execution_success and verification_success, + "execution": {"success": execution_success, "stage": "solver"}, + "solver": {"success": bool(result.success), "status": int(result.status)}, + "verification": {"success": verification_success, "metric": "manifold.Feas_eval"}, + "mathematical_conclusion": "not_assessed", "status": int(result.status), "message": str(result.message), "fval": float(result.fun), @@ -108,16 +134,20 @@ def obj_fun(x): }, } - results_dir = Path(args.results_dir) results_dir.mkdir(parents=True, exist_ok=True) + if results_dir.is_symlink(): + raise RuntimeError("results directory became a symlink") out_path = results_dir / "solver_summary.json" + if out_path.is_symlink(): + raise RuntimeError("refusing to replace symlinked solver_summary.json") out_path.write_text(json.dumps(summary, indent=2, sort_keys=True)) print(json.dumps(summary, indent=2, sort_keys=True)) print(f"wrote {out_path}") + return 0 if summary["success"] else 1 if __name__ == "__main__": - main() + raise SystemExit(main()) ''' @@ -131,8 +161,12 @@ def main(): args = parser.parse_args() output_dir = Path(args.output_dir) + if output_dir.is_symlink() or (output_dir.exists() and not output_dir.is_dir()): + parser.error("--output-dir must be a real directory") output_dir.mkdir(parents=True, exist_ok=True) runner_path = output_dir / "run_dictionary_learning.py" + if runner_path.is_symlink() or (runner_path.exists() and not runner_path.is_file()): + parser.error("refusing to replace an unsafe generated runner path") runner_path.write_text(textwrap.dedent(RUNNER)) runner_path.chmod(0o755) print(runner_path) diff --git a/tests/test_cdopt_runner_generators.py b/tests/test_cdopt_runner_generators.py new file mode 100644 index 0000000..a0cc55b --- /dev/null +++ b/tests/test_cdopt_runner_generators.py @@ -0,0 +1,59 @@ +from __future__ import annotations + +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[1] +SCRIPTS = ROOT / "skills" / "cdopt-optimization" / "scripts" +GENERATORS = ( + ("write_constrained_layer_runner.py", "run_constrained_layer.py", "--steps"), + ("write_constrained_rnn_runner.py", "run_constrained_rnn.py", "--steps"), + ("write_stiefel_dictionary_runner.py", "run_dictionary_learning.py", "--maxiter"), +) + + +class CdoptRunnerGeneratorTests(unittest.TestCase): + def test_generators_write_compilable_runners_with_failure_exit(self) -> None: + for generator, runner_name, _bounded_option in GENERATORS: + with self.subTest(generator=generator), tempfile.TemporaryDirectory() as directory: + output = Path(directory) / "generated" + result = subprocess.run( + [sys.executable, str(SCRIPTS / generator), "--output-dir", str(output)], + capture_output=True, + text=True, + check=False, + ) + self.assertEqual(result.returncode, 0, result.stderr) + runner = output / runner_name + subprocess.run( + [sys.executable, "-m", "py_compile", str(runner)], check=True + ) + text = runner.read_text(encoding="utf-8") + self.assertIn("raise SystemExit(main())", text) + self.assertIn('return 0 if summary["success"] else 1', text) + + @unittest.skipUnless(hasattr(Path, "symlink_to"), "symlinks unavailable") + def test_generators_reject_symlinked_output_runner(self) -> None: + for generator, runner_name, _bounded_option in GENERATORS: + with self.subTest(generator=generator), tempfile.TemporaryDirectory() as directory: + output = Path(directory) / "generated" + output.mkdir() + target = Path(directory) / "victim.py" + target.write_text("unchanged", encoding="utf-8") + (output / runner_name).symlink_to(target) + result = subprocess.run( + [sys.executable, str(SCRIPTS / generator), "--output-dir", str(output)], + capture_output=True, + text=True, + check=False, + ) + self.assertNotEqual(result.returncode, 0) + self.assertEqual(target.read_text(encoding="utf-8"), "unchanged") + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file