Skills › Scientific Agent Skills › hypogenic › plan_run.py
plan_run.py
249 lines · skills/hypogenic/scripts/plan_run.py at 5f2bc03
1#!/usr/bin/env python3
2"""Create a deterministic token/cost preflight without calling any model."""
4from __future__ import annotations
6import argparse
7from collections.abc import Sequence
8from decimal import Decimal, ROUND_HALF_UP
10if __package__:
11 from ._common import (
12 HYPOGENIC_COMMIT,
13 HYPOGENIC_VERSION,
14 HYPOGENIC_WHEEL_SHA256,
15 MAX_CONFIG_BYTES,
16 CliError,
17 checked_input_file,
18 emit_error,
19 emit_json,
20 load_structured_document,
21 named_env_presence,
22 sha256_file,
23 validate_run_config,
24 )
25else:
26 from _common import ( # type: ignore
27 HYPOGENIC_COMMIT,
28 HYPOGENIC_VERSION,
29 HYPOGENIC_WHEEL_SHA256,
30 MAX_CONFIG_BYTES,
31 CliError,
32 checked_input_file,
33 emit_error,
34 emit_json,
35 load_structured_document,
36 named_env_presence,
37 sha256_file,
38 validate_run_config,
39 )
42def build_parser() -> argparse.ArgumentParser:
43 parser = argparse.ArgumentParser(
44 description=(
45 "Generate a plan-only HypoGeniC token/cost upper bound. No package, "
46 "model, provider SDK, dataset, .env, or network service is loaded."
47 )
48 )
49 parser.add_argument("--config", required=True, help="Local reviewed run policy")
50 parser.add_argument("--root", default=".", help="Existing local I/O boundary")
51 parser.add_argument(
52 "--check-env",
53 action="store_true",
54 help="Check only the configured named credential variable; never print its value",
55 )
56 parser.add_argument(
57 "--check-inputs",
58 action="store_true",
59 help="Require the configured task config and dataset manifest to exist",
60 )
61 return parser
64def _money(value: Decimal) -> float:
65 return float(value.quantize(Decimal("0.000001"), rounding=ROUND_HALF_UP))
68def make_plan(
69 config: dict,
70 *,
71 config_sha256: str,
72 check_env: bool,
73 inputs_checked: bool,
74) -> dict:
75 limits = config["limits"]
76 pricing = config["pricing"]
77 max_requests = limits["max_requests"]
78 input_tokens = max_requests * limits["max_input_tokens_per_request"]
79 output_tokens = max_requests * limits["max_output_tokens_per_request"]
80 total_tokens = input_tokens + output_tokens
81 token_cap_ok = total_tokens <= limits["max_total_tokens"]
83 rates_available = pricing["input_usd_per_million_tokens"] is not None
84 estimated_cost: float | None = None
85 cost_cap_ok = False
86 if rates_available:
87 million = Decimal(1_000_000)
88 input_cost = (
89 Decimal(input_tokens)
90 * Decimal(str(pricing["input_usd_per_million_tokens"]))
91 / million
92 )
93 output_cost = (
94 Decimal(output_tokens)
95 * Decimal(str(pricing["output_usd_per_million_tokens"]))
96 / million
97 )
98 exact_cost = input_cost + output_cost
99 estimated_cost = _money(exact_cost)
100 cost_cap_ok = exact_cost <= Decimal(
101 str(limits["max_cost_usd"])
102 )
104 environment = (
105 named_env_presence(config)
106 if check_env
107 else {
108 "checked": False,
109 "name": config["provider"]["credential_env"],
110 "value_included": False,
111 }
112 )
113 environment_ready = (
114 not check_env
115 or environment.get("required") is False
116 or environment.get("present") is True
117 )
118 ready = (
119 rates_available
120 and token_cap_ok
121 and cost_cap_ok
122 and environment_ready
123 and config["execution"]["mode"] == "plan_only"
124 )
125 blockers: list[str] = []
126 if not rates_available:
127 blockers.append("current provider token prices are not supplied")
128 if not token_cap_ok:
129 blockers.append("worst-case tokens exceed max_total_tokens")
130 if rates_available and not cost_cap_ok:
131 blockers.append("worst-case API cost exceeds max_cost_usd")
132 if not environment_ready:
133 blockers.append("the named provider credential is absent")
135 return {
136 "ok": ready,
137 "plan_kind": "hypogenic_preflight_only",
138 "schema_version": config["schema_version"],
139 "config_sha256": config_sha256,
140 "software": {
141 "package": "hypogenic",
142 "version": HYPOGENIC_VERSION,
143 "source_commit": HYPOGENIC_COMMIT,
144 "wheel_sha256": HYPOGENIC_WHEEL_SHA256,
145 "package_imported": False,
146 },
147 "provider": {
148 "type": config["provider"]["type"],
149 "model": config["provider"]["model"],
150 "data_destination": config["provider"]["data_destination"],
151 "credential_environment": environment,
152 "local_model_path_configured": (
153 config["provider"]["local_model_path"] is not None
154 ),
155 },
156 "data": {
157 "task_config": config["data"]["task_config"],
158 "dataset_manifest": config["data"]["dataset_manifest"],
159 "output_directory": config["data"]["output_directory"],
160 "test_split_policy": "locked_until_final",
161 "test_split_sent_during_generation": False,
162 "inputs_checked": inputs_checked,
163 },
164 "limits": limits,
165 "cost_upper_bound": {
166 "method": (
167 "max_requests * per-request input/output token caps * "
168 "user-reviewed per-million-token rates"
169 ),
170 "maximum_input_tokens": input_tokens,
171 "maximum_output_tokens": output_tokens,
172 "maximum_total_tokens": total_tokens,
173 "within_total_token_cap": token_cap_ok,
174 "pricing_available": rates_available,
175 "pricing_reviewed_on": pricing["reviewed_on"],
176 "pricing_source": pricing["source"],
177 "estimated_maximum_cost_usd": estimated_cost,
178 "configured_cost_cap_usd": limits["max_cost_usd"],
179 "within_cost_cap": cost_cap_ok if rates_available else None,
180 "not_a_provider_quote": True,
181 },
182 "execution": {
183 "ready_for_separate_execution_review": ready,
184 "blockers": blockers,
185 "external_calls_authorized": False,
186 "requires_separate_confirmation": True,
187 "network_access": False,
188 "model_called": False,
189 "commands_executed": False,
190 },
191 "logging": {
192 "level": config["logging"]["level"],
193 "prompt_content_redacted": True,
194 "response_content_redacted": True,
195 "credential_values_included": False,
196 },
197 "caveats": [
198 "This is a conservative arithmetic bound, not tokenizer output or a provider quote.",
199 "The pinned upstream CLI does not enforce this dollar budget.",
200 "This plan does not verify model compatibility or the SDK endpoint actually used.",
201 "Verify current model limits, prices, account limits, and retention before execution.",
202 "Candidate hypotheses and benchmark metrics are not scientific validation.",
203 ],
204 }
207def main(argv: Sequence[str] | None = None) -> int:
208 args = build_parser().parse_args(argv)
209 try:
210 config_path = checked_input_file(
211 args.config,
212 root=args.root,
213 suffixes={".json", ".yaml", ".yml"},
214 max_bytes=MAX_CONFIG_BYTES,
215 )
216 config = validate_run_config(
217 load_structured_document(
218 args.config,
219 root=args.root,
220 max_bytes=MAX_CONFIG_BYTES,
221 )
222 )
223 if args.check_inputs:
224 checked_input_file(
225 config["data"]["task_config"],
226 root=args.root,
227 suffixes={".json", ".yaml", ".yml"},
228 max_bytes=MAX_CONFIG_BYTES,
229 )
230 checked_input_file(
231 config["data"]["dataset_manifest"],
232 root=args.root,
233 suffixes={".json"},
234 max_bytes=MAX_CONFIG_BYTES,
235 )
236 plan = make_plan(
237 config,
238 config_sha256=sha256_file(config_path, max_bytes=MAX_CONFIG_BYTES),
239 check_env=args.check_env,
240 inputs_checked=bool(args.check_inputs),
241 )
242 emit_json(plan)
243 return 0 if plan["ok"] else 3
244 except (CliError, OSError) as error:
245 return emit_error(error)
248if __name__ == "__main__":
249 raise SystemExit(main())