| 1 | #!/usr/bin/env python3 |
| 2 | """Hermetic tests for scripts/check-runtime-contract-budget.py.""" |
| 3 | |
| 4 | from __future__ import annotations |
| 5 | |
| 6 | import importlib.util |
| 7 | import io |
| 8 | import json |
| 9 | import os |
| 10 | import stat |
| 11 | import sys |
| 12 | import tempfile |
| 13 | import unittest |
| 14 | from contextlib import redirect_stderr, redirect_stdout |
| 15 | from pathlib import Path |
| 16 | from unittest import mock |
| 17 | |
| 18 | ROOT = Path(__file__).resolve().parents[1] |
| 19 | SCRIPT = ROOT / "scripts" / "check-runtime-contract-budget.py" |
| 20 | |
| 21 | SPEC = importlib.util.spec_from_file_location("check_runtime_contract_budget", SCRIPT) |
| 22 | assert SPEC and SPEC.loader |
| 23 | mod = importlib.util.module_from_spec(SPEC) |
| 24 | sys.modules[SPEC.name] = mod |
| 25 | SPEC.loader.exec_module(mod) |
| 26 | |
| 27 | |
| 28 | def receipt_fixture() -> dict: |
| 29 | def stage(stage_id: str, byte_count: int, delta: int | None = None) -> dict: |
| 30 | value = { |
| 31 | "bytes": byte_count, |
| 32 | "identity_sha256": mod.hashlib.sha256(stage_id.encode()).hexdigest(), |
| 33 | } |
| 34 | if delta is not None: |
| 35 | value["delta_bytes"] = delta |
| 36 | return value |
| 37 | |
| 38 | def tool_surface(names: list[str], byte_count: int) -> dict: |
| 39 | names = sorted(names) |
| 40 | return { |
| 41 | "bytes": byte_count, |
| 42 | "identity_sha256": mod.tool_identity_digest(names), |
| 43 | "tokens_est": (byte_count + 3) // 4, |
| 44 | "tool_names": names, |
| 45 | "tools": len(names), |
| 46 | } |
| 47 | |
| 48 | return { |
| 49 | "document_kind": mod.RECEIPT_KIND, |
| 50 | "schema_version": mod.SCHEMA_VERSION, |
| 51 | "representative_context": { |
| 52 | "fixture_id": mod.REPRESENTATIVE_FIXTURE_ID, |
| 53 | "prompts_byte_identical": True, |
| 54 | "stages": { |
| 55 | "base": stage("base", 24000), |
| 56 | "project": stage("project", 24100, 100), |
| 57 | "instructions": stage("instructions", 24220, 120), |
| 58 | "skill": stage("skill", 24360, 140), |
| 59 | "memory": stage("memory", 24450, 90), |
| 60 | "goal": stage("goal", 24540, 90), |
| 61 | "handoff": stage("handoff", 24780, 240), |
| 62 | }, |
| 63 | "system_prompt_blocks": 6, |
| 64 | "total_bytes": 24780, |
| 65 | "total_tokens_est": 6195, |
| 66 | }, |
| 67 | "skill_discovery": { |
| 68 | "first_delta": { |
| 69 | "directories_visited": 1, |
| 70 | "root_discovery_calls": 1, |
| 71 | "skill_md_read_attempts": 1, |
| 72 | }, |
| 73 | "prompts_byte_identical": True, |
| 74 | "second_delta": { |
| 75 | "directories_visited": 1, |
| 76 | "root_discovery_calls": 1, |
| 77 | "skill_md_read_attempts": 1, |
| 78 | }, |
| 79 | }, |
| 80 | "system_prompt": { |
| 81 | "modes": { |
| 82 | "plan": { |
| 83 | "system_prompt_bytes": 24000, |
| 84 | "system_prompt_tokens_est": 6000, |
| 85 | "system_prompt_blocks": 4, |
| 86 | "mode_instructions_bytes": 700, |
| 87 | "mode_instructions_tokens_est": 175, |
| 88 | }, |
| 89 | "act": { |
| 90 | "system_prompt_bytes": 25040, |
| 91 | "system_prompt_tokens_est": 6260, |
| 92 | "system_prompt_blocks": 4, |
| 93 | "mode_instructions_bytes": 805, |
| 94 | "mode_instructions_tokens_est": 202, |
| 95 | }, |
| 96 | "operate": { |
| 97 | "system_prompt_bytes": 24500, |
| 98 | "system_prompt_tokens_est": 6125, |
| 99 | "system_prompt_blocks": 4, |
| 100 | "mode_instructions_bytes": 750, |
| 101 | "mode_instructions_tokens_est": 188, |
| 102 | }, |
| 103 | } |
| 104 | }, |
| 105 | "tool_catalog": { |
| 106 | "surface_profile": mod.TOOL_SURFACE_PROFILE, |
| 107 | "execution_shell": "bash", |
| 108 | "modes": { |
| 109 | "plan": { |
| 110 | "full": tool_surface(["File", "Git", "create_goal"], 18000), |
| 111 | "active": tool_surface(["File", "Git"], 12000), |
| 112 | }, |
| 113 | "act": { |
| 114 | "full": tool_surface(["Bash", "File", "verify"], 20000), |
| 115 | "active": tool_surface(["Bash", "File"], 13000), |
| 116 | }, |
| 117 | "operate": { |
| 118 | "full": tool_surface(["File", "Run", "verify"], 21000), |
| 119 | "active": tool_surface(["File", "Run"], 14000), |
| 120 | }, |
| 121 | }, |
| 122 | }, |
| 123 | } |
| 124 | |
| 125 | |
| 126 | def set_path(document: dict, path: tuple[str, ...], value: int) -> None: |
| 127 | target = document |
| 128 | for part in path[:-1]: |
| 129 | target = target[part] |
| 130 | target[path[-1]] = value |
| 131 | |
| 132 | |
| 133 | def write_documents(tmp: str, receipt: dict, budget: dict) -> tuple[Path, Path]: |
| 134 | receipt_path = Path(tmp) / "receipt.json" |
| 135 | budget_path = Path(tmp) / "budget.json" |
| 136 | receipt_path.write_text(json.dumps(receipt), encoding="utf-8") |
| 137 | budget_path.write_text(json.dumps(budget), encoding="utf-8") |
| 138 | return receipt_path, budget_path |
| 139 | |
| 140 | |
| 141 | class RuntimeContractBudgetTests(unittest.TestCase): |
| 142 | def test_equal_fixture_passes_all_ceiling_checks(self) -> None: |
| 143 | receipt = receipt_fixture() |
| 144 | budget = mod.budget_from_receipt(receipt) |
| 145 | self.assertEqual(len(mod.METRICS), 55) |
| 146 | self.assertEqual(mod.compare(receipt, budget), ([], [])) |
| 147 | |
| 148 | def test_unpinned_shell_receipt_is_rejected(self) -> None: |
| 149 | receipt = receipt_fixture() |
| 150 | budget = mod.budget_from_receipt(receipt) |
| 151 | del receipt["tool_catalog"]["execution_shell"] |
| 152 | with self.assertRaisesRegex(mod.RuntimeContractError, "execution_shell"): |
| 153 | mod.compare(receipt, budget) |
| 154 | |
| 155 | def test_wrong_shell_is_rejected_even_with_matching_budget(self) -> None: |
| 156 | receipt = receipt_fixture() |
| 157 | budget = mod.budget_from_receipt(receipt) |
| 158 | for shell in ["/bin/bash", "/bin/zsh", "pwsh", None]: |
| 159 | with self.subTest(shell=shell): |
| 160 | receipt["tool_catalog"]["execution_shell"] = shell |
| 161 | budget["tool_catalog"]["execution_shell"] = shell |
| 162 | with self.assertRaisesRegex(mod.RuntimeContractError, "execution_shell"): |
| 163 | mod.compare(receipt, budget) |
| 164 | |
| 165 | def test_every_owned_metric_rejects_an_increase(self) -> None: |
| 166 | budget = mod.budget_from_receipt(receipt_fixture()) |
| 167 | for path, _label in mod.METRICS: |
| 168 | with self.subTest(metric=".".join(path)): |
| 169 | receipt = receipt_fixture() |
| 170 | current = mod.metric_value(receipt, path, "receipt") |
| 171 | set_path(receipt, path, current + 1) |
| 172 | if path[-1] == "tools": |
| 173 | with self.assertRaisesRegex( |
| 174 | mod.RuntimeContractError, "tool_names length" |
| 175 | ): |
| 176 | mod.compare(receipt, budget) |
| 177 | continue |
| 178 | increases, decreases = mod.compare(receipt, budget) |
| 179 | self.assertEqual([item[0] for item in increases], [".".join(path)]) |
| 180 | self.assertEqual(decreases, []) |
| 181 | |
| 182 | def test_decrease_passes_and_is_reported_for_tightening(self) -> None: |
| 183 | receipt = receipt_fixture() |
| 184 | budget = mod.budget_from_receipt(receipt) |
| 185 | path = ("system_prompt", "modes", "operate", "system_prompt_bytes") |
| 186 | set_path(receipt, path, 24400) |
| 187 | increases, decreases = mod.compare(receipt, budget) |
| 188 | self.assertEqual(increases, []) |
| 189 | self.assertEqual([item[0] for item in decreases], [".".join(path)]) |
| 190 | |
| 191 | def test_missing_and_non_integer_metrics_are_rejected(self) -> None: |
| 192 | receipt = receipt_fixture() |
| 193 | budget = mod.budget_from_receipt(receipt) |
| 194 | del receipt["tool_catalog"]["modes"]["act"]["active"]["bytes"] |
| 195 | with self.assertRaisesRegex(mod.RuntimeContractError, "active.bytes"): |
| 196 | mod.compare(receipt, budget) |
| 197 | |
| 198 | receipt = receipt_fixture() |
| 199 | receipt["tool_catalog"]["modes"]["act"]["active"]["bytes"] = True |
| 200 | with self.assertRaisesRegex(mod.RuntimeContractError, "non-negative integer"): |
| 201 | mod.compare(receipt, budget) |
| 202 | |
| 203 | def test_document_kinds_and_schema_versions_are_distinct_and_required(self) -> None: |
| 204 | receipt = receipt_fixture() |
| 205 | budget = mod.budget_from_receipt(receipt) |
| 206 | with self.assertRaisesRegex(mod.RuntimeContractError, "document_kind"): |
| 207 | mod.compare(budget, budget) |
| 208 | |
| 209 | receipt.pop("schema_version") |
| 210 | with self.assertRaisesRegex(mod.RuntimeContractError, "schema_version"): |
| 211 | mod.compare(receipt, budget) |
| 212 | |
| 213 | receipt = receipt_fixture() |
| 214 | budget["schema_version"] = mod.SCHEMA_VERSION + 1 |
| 215 | with self.assertRaisesRegex(mod.RuntimeContractError, "schema_version"): |
| 216 | mod.compare(receipt, budget) |
| 217 | |
| 218 | def test_saved_receipt_requires_byte_identical_prompts(self) -> None: |
| 219 | receipt = receipt_fixture() |
| 220 | budget = mod.budget_from_receipt(receipt) |
| 221 | receipt["skill_discovery"]["prompts_byte_identical"] = False |
| 222 | with self.assertRaisesRegex(mod.RuntimeContractError, "must be true"): |
| 223 | mod.compare(receipt, budget) |
| 224 | |
| 225 | def test_representative_fixture_identity_and_stability_are_required(self) -> None: |
| 226 | receipt = receipt_fixture() |
| 227 | budget = mod.budget_from_receipt(receipt) |
| 228 | receipt["representative_context"]["fixture_id"] = "ambient-developer-state" |
| 229 | with self.assertRaisesRegex(mod.RuntimeContractError, "fixture_id"): |
| 230 | mod.compare(receipt, budget) |
| 231 | |
| 232 | receipt = receipt_fixture() |
| 233 | receipt["representative_context"]["prompts_byte_identical"] = False |
| 234 | with self.assertRaisesRegex(mod.RuntimeContractError, "must be true"): |
| 235 | mod.compare(receipt, budget) |
| 236 | |
| 237 | def test_tool_identity_rejects_equal_size_substitution_and_removal(self) -> None: |
| 238 | budget = mod.budget_from_receipt(receipt_fixture()) |
| 239 | |
| 240 | receipt = receipt_fixture() |
| 241 | active = receipt["tool_catalog"]["modes"]["act"]["active"] |
| 242 | active["tool_names"] = sorted(["File", "Hash"]) |
| 243 | active["identity_sha256"] = mod.tool_identity_digest(active["tool_names"]) |
| 244 | with self.assertRaisesRegex(mod.RuntimeContractError, "identity changed"): |
| 245 | mod.compare(receipt, budget) |
| 246 | |
| 247 | receipt = receipt_fixture() |
| 248 | full = receipt["tool_catalog"]["modes"]["plan"]["full"] |
| 249 | full["tool_names"].remove("Git") |
| 250 | full["tools"] -= 1 |
| 251 | full["identity_sha256"] = mod.tool_identity_digest(full["tool_names"]) |
| 252 | with self.assertRaisesRegex(mod.RuntimeContractError, "identity changed"): |
| 253 | mod.compare(receipt, budget) |
| 254 | |
| 255 | def test_tool_identity_rejects_missing_structure_and_wrong_mode_swap(self) -> None: |
| 256 | budget = mod.budget_from_receipt(receipt_fixture()) |
| 257 | |
| 258 | receipt = receipt_fixture() |
| 259 | del receipt["tool_catalog"]["modes"]["operate"]["full"]["tool_names"] |
| 260 | with self.assertRaisesRegex(mod.RuntimeContractError, "missing required field"): |
| 261 | mod.compare(receipt, budget) |
| 262 | |
| 263 | receipt = receipt_fixture() |
| 264 | plan = receipt["tool_catalog"]["modes"]["plan"]["active"] |
| 265 | act = receipt["tool_catalog"]["modes"]["act"]["active"] |
| 266 | for field in ["tool_names", "identity_sha256"]: |
| 267 | plan[field], act[field] = act[field], plan[field] |
| 268 | with self.assertRaisesRegex(mod.RuntimeContractError, "identity changed"): |
| 269 | mod.compare(receipt, budget) |
| 270 | |
| 271 | def test_representative_stage_identity_rejects_same_length_substitution(self) -> None: |
| 272 | receipt = receipt_fixture() |
| 273 | budget = mod.budget_from_receipt(receipt) |
| 274 | digest = receipt["representative_context"]["stages"]["project"][ |
| 275 | "identity_sha256" |
| 276 | ] |
| 277 | receipt["representative_context"]["stages"]["project"][ |
| 278 | "identity_sha256" |
| 279 | ] = ("0" if digest[0] != "0" else "1") + digest[1:] |
| 280 | with self.assertRaisesRegex(mod.RuntimeContractError, "identity changed"): |
| 281 | mod.compare(receipt, budget) |
| 282 | |
| 283 | def test_same_resolved_receipt_and_budget_path_is_rejected(self) -> None: |
| 284 | budget = mod.budget_from_receipt(receipt_fixture()) |
| 285 | with tempfile.TemporaryDirectory() as tmp: |
| 286 | budget_path = Path(tmp) / "budget.json" |
| 287 | nested = Path(tmp) / "nested" |
| 288 | nested.mkdir() |
| 289 | budget_path.write_text(json.dumps(budget), encoding="utf-8") |
| 290 | alias = nested / ".." / "budget.json" |
| 291 | errors = io.StringIO() |
| 292 | with redirect_stderr(errors): |
| 293 | result = mod.main( |
| 294 | ["--receipt", str(alias), "--budget", str(budget_path)] |
| 295 | ) |
| 296 | self.assertEqual(result, 2) |
| 297 | self.assertIn("distinct filesystem paths", errors.getvalue()) |
| 298 | |
| 299 | def test_receipt_path_avoids_running_measurement(self) -> None: |
| 300 | receipt = receipt_fixture() |
| 301 | budget = mod.budget_from_receipt(receipt) |
| 302 | with tempfile.TemporaryDirectory() as tmp: |
| 303 | receipt_path, budget_path = write_documents(tmp, receipt, budget) |
| 304 | output = io.StringIO() |
| 305 | with ( |
| 306 | mock.patch.object( |
| 307 | mod, "run_measurement", side_effect=AssertionError("must stay hermetic") |
| 308 | ), |
| 309 | redirect_stdout(output), |
| 310 | ): |
| 311 | result = mod.main( |
| 312 | ["--receipt", str(receipt_path), "--budget", str(budget_path)] |
| 313 | ) |
| 314 | self.assertEqual(result, 0) |
| 315 | self.assertIn("all 55 metrics are exactly at budget", output.getvalue()) |
| 316 | |
| 317 | def test_default_measurement_forces_cargo_offline(self) -> None: |
| 318 | receipt = receipt_fixture() |
| 319 | completed = mock.Mock(returncode=0, stdout=json.dumps(receipt), stderr="") |
| 320 | with mock.patch.object(mod.subprocess, "run", return_value=completed) as run: |
| 321 | measured = mod.run_measurement() |
| 322 | self.assertEqual(measured, receipt) |
| 323 | command = run.call_args.args[0] |
| 324 | environment = run.call_args.kwargs["env"] |
| 325 | self.assertEqual(command, [sys.executable, str(mod.MEASURE_SCRIPT)]) |
| 326 | self.assertEqual(environment["CARGO_NET_OFFLINE"], "true") |
| 327 | |
| 328 | def test_cli_decrease_prints_a_tightening_command(self) -> None: |
| 329 | receipt = receipt_fixture() |
| 330 | budget = mod.budget_from_receipt(receipt) |
| 331 | set_path(receipt, ("system_prompt", "modes", "plan", "system_prompt_bytes"), 23000) |
| 332 | with tempfile.TemporaryDirectory() as tmp: |
| 333 | receipt_path, budget_path = write_documents(tmp, receipt, budget) |
| 334 | output = io.StringIO() |
| 335 | with redirect_stdout(output): |
| 336 | result = mod.main( |
| 337 | ["--receipt", str(receipt_path), "--budget", str(budget_path)] |
| 338 | ) |
| 339 | self.assertEqual(result, 0) |
| 340 | self.assertIn("1 can be tightened", output.getvalue()) |
| 341 | self.assertIn("--update", output.getvalue()) |
| 342 | |
| 343 | def test_update_tightens_decreases_and_preserves_permissions(self) -> None: |
| 344 | receipt = receipt_fixture() |
| 345 | budget = mod.budget_from_receipt(receipt) |
| 346 | path = ("tool_catalog", "modes", "plan", "active", "bytes") |
| 347 | set_path(receipt, path, 11000) |
| 348 | with tempfile.TemporaryDirectory() as tmp: |
| 349 | receipt_path, budget_path = write_documents(tmp, receipt, budget) |
| 350 | os.chmod(budget_path, 0o640) |
| 351 | with redirect_stdout(io.StringIO()): |
| 352 | result = mod.main( |
| 353 | [ |
| 354 | "--receipt", |
| 355 | str(receipt_path), |
| 356 | "--budget", |
| 357 | str(budget_path), |
| 358 | "--update", |
| 359 | ] |
| 360 | ) |
| 361 | updated = json.loads(budget_path.read_text(encoding="utf-8")) |
| 362 | updated_mode = stat.S_IMODE(budget_path.stat().st_mode) |
| 363 | self.assertEqual(result, 0) |
| 364 | self.assertEqual(mod.metric_value(updated, path, "budget"), 11000) |
| 365 | self.assertEqual(updated_mode, 0o640) |
| 366 | |
| 367 | def test_atomic_update_failure_leaves_original_and_removes_temporary_file(self) -> None: |
| 368 | budget = mod.budget_from_receipt(receipt_fixture()) |
| 369 | replacement = mod.budget_from_receipt(receipt_fixture()) |
| 370 | replacement["_comment"] = "replacement" |
| 371 | with tempfile.TemporaryDirectory() as tmp: |
| 372 | budget_path = Path(tmp) / "budget.json" |
| 373 | original = json.dumps(budget) |
| 374 | budget_path.write_text(original, encoding="utf-8") |
| 375 | with ( |
| 376 | mock.patch.object(mod.os, "replace", side_effect=OSError("stop")), |
| 377 | self.assertRaisesRegex(OSError, "stop"), |
| 378 | ): |
| 379 | mod.write_budget_atomic(budget_path, replacement) |
| 380 | after = budget_path.read_text(encoding="utf-8") |
| 381 | temporary_files = list(Path(tmp).glob(".budget.json.*.tmp")) |
| 382 | self.assertEqual(after, original) |
| 383 | self.assertEqual(temporary_files, []) |
| 384 | |
| 385 | def test_update_refuses_an_increase_without_rewriting_budget(self) -> None: |
| 386 | receipt = receipt_fixture() |
| 387 | budget = mod.budget_from_receipt(receipt) |
| 388 | path = ("skill_discovery", "second_delta", "directories_visited") |
| 389 | set_path(receipt, path, 2) |
| 390 | with tempfile.TemporaryDirectory() as tmp: |
| 391 | receipt_path, budget_path = write_documents(tmp, receipt, budget) |
| 392 | original = budget_path.read_text(encoding="utf-8") |
| 393 | with redirect_stderr(io.StringIO()): |
| 394 | result = mod.main( |
| 395 | [ |
| 396 | "--receipt", |
| 397 | str(receipt_path), |
| 398 | "--budget", |
| 399 | str(budget_path), |
| 400 | "--update", |
| 401 | ] |
| 402 | ) |
| 403 | after = budget_path.read_text(encoding="utf-8") |
| 404 | self.assertEqual(result, 1) |
| 405 | self.assertEqual(after, original) |
| 406 | |
| 407 | |
| 408 | if __name__ == "__main__": |
| 409 | raise SystemExit(unittest.main()) |
| 410 |