Skills › Scientific Agent Skills › hypogenic › validate_config.py
validate_config.py
192 lines · skills/hypogenic/scripts/validate_config.py at 5f2bc03
1#!/usr/bin/env python3
2"""Validate HypoGeniC task configs and the skill's plan-only run policy."""
4from __future__ import annotations
6import argparse
7from collections.abc import Sequence
9if __package__:
10 from ._common import (
11 MAX_CONFIG_BYTES,
12 MAX_JSON_BYTES,
13 CliError,
14 checked_input_file,
15 emit_error,
16 emit_json,
17 load_structured_document,
18 named_env_presence,
19 validate_run_config,
20 validate_task_config,
21 )
22else:
23 from _common import ( # type: ignore
24 MAX_CONFIG_BYTES,
25 MAX_JSON_BYTES,
26 CliError,
27 checked_input_file,
28 emit_error,
29 emit_json,
30 load_structured_document,
31 named_env_presence,
32 validate_run_config,
33 validate_task_config,
34 )
37def build_parser() -> argparse.ArgumentParser:
38 parser = argparse.ArgumentParser(
39 description=(
40 "Validate strict local HypoGeniC configuration without importing "
41 "HypoGeniC, loading .env, enumerating the environment, or calling a model."
42 )
43 )
44 subparsers = parser.add_subparsers(dest="command", required=True)
46 run = subparsers.add_parser(
47 "run",
48 help="Validate the skill-local provider/budget/run policy",
49 )
50 run.add_argument("--input", required=True, help="Local .json/.yaml run policy")
51 run.add_argument("--root", default=".", help="Existing local I/O boundary")
52 run.add_argument(
53 "--check-env",
54 action="store_true",
55 help="Check only the validated provider-specific variable name; never print its value",
56 )
57 run.add_argument(
58 "--check-paths",
59 action="store_true",
60 help="Require task_config and dataset_manifest to exist inside root",
61 )
63 task = subparsers.add_parser(
64 "task",
65 help="Validate the pinned source's task-config surface",
66 )
67 task.add_argument("--input", required=True, help="Local .json/.yaml task config")
68 task.add_argument("--root", default=".", help="Existing local I/O boundary")
69 task.add_argument(
70 "--check-data-files",
71 action="store_true",
72 help="Require train/validation/test and optional OOD JSON beside the task config",
73 )
74 return parser
77def _run_policy(args: argparse.Namespace) -> int:
78 document = load_structured_document(
79 args.input,
80 root=args.root,
81 max_bytes=MAX_CONFIG_BYTES,
82 )
83 config = validate_run_config(document)
84 if args.check_paths:
85 checked_input_file(
86 config["data"]["task_config"],
87 root=args.root,
88 suffixes={".json", ".yaml", ".yml"},
89 max_bytes=MAX_CONFIG_BYTES,
90 )
91 checked_input_file(
92 config["data"]["dataset_manifest"],
93 root=args.root,
94 suffixes={".json"},
95 max_bytes=MAX_CONFIG_BYTES,
96 )
97 environment = (
98 named_env_presence(config)
99 if args.check_env
100 else {
101 "checked": False,
102 "name": config["provider"]["credential_env"],
103 "value_included": False,
104 }
105 )
106 emit_json(
107 {
108 "ok": True,
109 "schema": "hypogenic-local-run-policy",
110 "schema_version": config["schema_version"],
111 "provider": {
112 "type": config["provider"]["type"],
113 "model": config["provider"]["model"],
114 "data_destination": config["provider"]["data_destination"],
115 "local_model_path_configured": (
116 config["provider"]["local_model_path"] is not None
117 ),
118 },
119 "credential_environment": environment,
120 "paths_checked": bool(args.check_paths),
121 "external_calls_authorized": False,
122 "requires_separate_confirmation": True,
123 "send_test_split": False,
124 "redacted_logging_required": True,
125 "network_access": False,
126 "model_called": False,
127 "secret_values_included": False,
128 }
129 )
130 return 0
133def _task_config(args: argparse.Namespace) -> int:
134 config_path = checked_input_file(
135 args.input,
136 root=args.root,
137 suffixes={".json", ".yaml", ".yml"},
138 max_bytes=MAX_CONFIG_BYTES,
139 )
140 document = load_structured_document(
141 args.input,
142 root=args.root,
143 max_bytes=MAX_CONFIG_BYTES,
144 )
145 config = validate_task_config(document)
146 checked: list[str] = []
147 if args.check_data_files:
148 for key in (
149 "train_data_path",
150 "val_data_path",
151 "test_data_path",
152 "ood_data_path",
153 ):
154 if config.get(key) is None:
155 continue
156 checked_input_file(
157 config[key],
158 root=config_path.parent,
159 suffixes={".json"},
160 max_bytes=MAX_JSON_BYTES,
161 )
162 checked.append(key)
163 emit_json(
164 {
165 "ok": True,
166 "schema": "hypogenic-upstream-task-config",
167 "task_name": config["task_name"],
168 "label_name": config["label_name"],
169 "prompt_template_keys": config["prompt_template_keys"],
170 "prompt_text_interpreted": False,
171 "data_files_checked": checked,
172 "split_paths_distinct": True,
173 "network_access": False,
174 "model_called": False,
175 "secret_values_included": False,
176 }
177 )
178 return 0
181def main(argv: Sequence[str] | None = None) -> int:
182 args = build_parser().parse_args(argv)
183 try:
184 if args.command == "run":
185 return _run_policy(args)
186 return _task_config(args)
187 except (CliError, OSError) as error:
188 return emit_error(error)
191if __name__ == "__main__":
192 raise SystemExit(main())