Hmbown
67adf84697
Measure the provider-free model-facing contract through production-owned seams. Keep deterministic minimal prompts by visible mode, build each Plan/Act/Operate tool surface through the canonical turn registry and request planner with inert wiring, and record full and active catalog counts, bytes, token estimates, sorted names, and stable identities. Add a hermetic representative prompt fixture that cumulatively introduces project authority, configured instructions, one workspace skill, memory, goal, and handoff. Normalize temporary roots before hashing every stage, require stable repeated identities, prove each prior marker appears exactly once while future markers remain absent, and ratchet adjacent stage deltas plus the final envelope. Enforce 55 one-way numeric ceilings and exact structural identities, while retaining the observable repeated skill-discovery baseline and future cached-zero payload. Reject equal-size tool substitutions, removals, missing identity, cross-mode swaps, and representative-stage substitutions; keep atomic permission-preserving budget updates and the locked offline heavy-lint gate.
392 lines
16 KiB
Python
392 lines
16 KiB
Python
#!/usr/bin/env python3
|
|
"""Hermetic tests for scripts/check-runtime-contract-budget.py."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
import os
|
|
import stat
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from contextlib import redirect_stderr, redirect_stdout
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
SCRIPT = ROOT / "scripts" / "check-runtime-contract-budget.py"
|
|
|
|
SPEC = importlib.util.spec_from_file_location("check_runtime_contract_budget", SCRIPT)
|
|
assert SPEC and SPEC.loader
|
|
mod = importlib.util.module_from_spec(SPEC)
|
|
sys.modules[SPEC.name] = mod
|
|
SPEC.loader.exec_module(mod)
|
|
|
|
|
|
def receipt_fixture() -> dict:
|
|
def stage(stage_id: str, byte_count: int, delta: int | None = None) -> dict:
|
|
value = {
|
|
"bytes": byte_count,
|
|
"identity_sha256": mod.hashlib.sha256(stage_id.encode()).hexdigest(),
|
|
}
|
|
if delta is not None:
|
|
value["delta_bytes"] = delta
|
|
return value
|
|
|
|
def tool_surface(names: list[str], byte_count: int) -> dict:
|
|
names = sorted(names)
|
|
return {
|
|
"bytes": byte_count,
|
|
"identity_sha256": mod.tool_identity_digest(names),
|
|
"tokens_est": (byte_count + 3) // 4,
|
|
"tool_names": names,
|
|
"tools": len(names),
|
|
}
|
|
|
|
return {
|
|
"document_kind": mod.RECEIPT_KIND,
|
|
"schema_version": mod.SCHEMA_VERSION,
|
|
"representative_context": {
|
|
"fixture_id": mod.REPRESENTATIVE_FIXTURE_ID,
|
|
"prompts_byte_identical": True,
|
|
"stages": {
|
|
"base": stage("base", 24000),
|
|
"project": stage("project", 24100, 100),
|
|
"instructions": stage("instructions", 24220, 120),
|
|
"skill": stage("skill", 24360, 140),
|
|
"memory": stage("memory", 24450, 90),
|
|
"goal": stage("goal", 24540, 90),
|
|
"handoff": stage("handoff", 24780, 240),
|
|
},
|
|
"system_prompt_blocks": 6,
|
|
"total_bytes": 24780,
|
|
"total_tokens_est": 6195,
|
|
},
|
|
"skill_discovery": {
|
|
"first_delta": {
|
|
"directories_visited": 1,
|
|
"root_discovery_calls": 1,
|
|
"skill_md_read_attempts": 1,
|
|
},
|
|
"prompts_byte_identical": True,
|
|
"second_delta": {
|
|
"directories_visited": 1,
|
|
"root_discovery_calls": 1,
|
|
"skill_md_read_attempts": 1,
|
|
},
|
|
},
|
|
"system_prompt": {
|
|
"modes": {
|
|
"plan": {
|
|
"system_prompt_bytes": 24000,
|
|
"system_prompt_tokens_est": 6000,
|
|
"system_prompt_blocks": 4,
|
|
"mode_instructions_bytes": 700,
|
|
"mode_instructions_tokens_est": 175,
|
|
},
|
|
"act": {
|
|
"system_prompt_bytes": 25040,
|
|
"system_prompt_tokens_est": 6260,
|
|
"system_prompt_blocks": 4,
|
|
"mode_instructions_bytes": 805,
|
|
"mode_instructions_tokens_est": 202,
|
|
},
|
|
"operate": {
|
|
"system_prompt_bytes": 24500,
|
|
"system_prompt_tokens_est": 6125,
|
|
"system_prompt_blocks": 4,
|
|
"mode_instructions_bytes": 750,
|
|
"mode_instructions_tokens_est": 188,
|
|
},
|
|
}
|
|
},
|
|
"tool_catalog": {
|
|
"surface_profile": mod.TOOL_SURFACE_PROFILE,
|
|
"modes": {
|
|
"plan": {
|
|
"full": tool_surface(["File", "Git", "create_goal"], 18000),
|
|
"active": tool_surface(["File", "Git"], 12000),
|
|
},
|
|
"act": {
|
|
"full": tool_surface(["Bash", "File", "verify"], 20000),
|
|
"active": tool_surface(["Bash", "File"], 13000),
|
|
},
|
|
"operate": {
|
|
"full": tool_surface(["File", "Run", "verify"], 21000),
|
|
"active": tool_surface(["File", "Run"], 14000),
|
|
},
|
|
},
|
|
},
|
|
}
|
|
|
|
|
|
def set_path(document: dict, path: tuple[str, ...], value: int) -> None:
|
|
target = document
|
|
for part in path[:-1]:
|
|
target = target[part]
|
|
target[path[-1]] = value
|
|
|
|
|
|
def write_documents(tmp: str, receipt: dict, budget: dict) -> tuple[Path, Path]:
|
|
receipt_path = Path(tmp) / "receipt.json"
|
|
budget_path = Path(tmp) / "budget.json"
|
|
receipt_path.write_text(json.dumps(receipt), encoding="utf-8")
|
|
budget_path.write_text(json.dumps(budget), encoding="utf-8")
|
|
return receipt_path, budget_path
|
|
|
|
|
|
class RuntimeContractBudgetTests(unittest.TestCase):
|
|
def test_equal_fixture_passes_all_ceiling_checks(self) -> None:
|
|
receipt = receipt_fixture()
|
|
budget = mod.budget_from_receipt(receipt)
|
|
self.assertEqual(len(mod.METRICS), 55)
|
|
self.assertEqual(mod.compare(receipt, budget), ([], []))
|
|
|
|
def test_every_owned_metric_rejects_an_increase(self) -> None:
|
|
budget = mod.budget_from_receipt(receipt_fixture())
|
|
for path, _label in mod.METRICS:
|
|
with self.subTest(metric=".".join(path)):
|
|
receipt = receipt_fixture()
|
|
current = mod.metric_value(receipt, path, "receipt")
|
|
set_path(receipt, path, current + 1)
|
|
if path[-1] == "tools":
|
|
with self.assertRaisesRegex(
|
|
mod.RuntimeContractError, "tool_names length"
|
|
):
|
|
mod.compare(receipt, budget)
|
|
continue
|
|
increases, decreases = mod.compare(receipt, budget)
|
|
self.assertEqual([item[0] for item in increases], [".".join(path)])
|
|
self.assertEqual(decreases, [])
|
|
|
|
def test_decrease_passes_and_is_reported_for_tightening(self) -> None:
|
|
receipt = receipt_fixture()
|
|
budget = mod.budget_from_receipt(receipt)
|
|
path = ("system_prompt", "modes", "operate", "system_prompt_bytes")
|
|
set_path(receipt, path, 24400)
|
|
increases, decreases = mod.compare(receipt, budget)
|
|
self.assertEqual(increases, [])
|
|
self.assertEqual([item[0] for item in decreases], [".".join(path)])
|
|
|
|
def test_missing_and_non_integer_metrics_are_rejected(self) -> None:
|
|
receipt = receipt_fixture()
|
|
budget = mod.budget_from_receipt(receipt)
|
|
del receipt["tool_catalog"]["modes"]["act"]["active"]["bytes"]
|
|
with self.assertRaisesRegex(mod.RuntimeContractError, "active.bytes"):
|
|
mod.compare(receipt, budget)
|
|
|
|
receipt = receipt_fixture()
|
|
receipt["tool_catalog"]["modes"]["act"]["active"]["bytes"] = True
|
|
with self.assertRaisesRegex(mod.RuntimeContractError, "non-negative integer"):
|
|
mod.compare(receipt, budget)
|
|
|
|
def test_document_kinds_and_schema_versions_are_distinct_and_required(self) -> None:
|
|
receipt = receipt_fixture()
|
|
budget = mod.budget_from_receipt(receipt)
|
|
with self.assertRaisesRegex(mod.RuntimeContractError, "document_kind"):
|
|
mod.compare(budget, budget)
|
|
|
|
receipt.pop("schema_version")
|
|
with self.assertRaisesRegex(mod.RuntimeContractError, "schema_version"):
|
|
mod.compare(receipt, budget)
|
|
|
|
receipt = receipt_fixture()
|
|
budget["schema_version"] = mod.SCHEMA_VERSION + 1
|
|
with self.assertRaisesRegex(mod.RuntimeContractError, "schema_version"):
|
|
mod.compare(receipt, budget)
|
|
|
|
def test_saved_receipt_requires_byte_identical_prompts(self) -> None:
|
|
receipt = receipt_fixture()
|
|
budget = mod.budget_from_receipt(receipt)
|
|
receipt["skill_discovery"]["prompts_byte_identical"] = False
|
|
with self.assertRaisesRegex(mod.RuntimeContractError, "must be true"):
|
|
mod.compare(receipt, budget)
|
|
|
|
def test_representative_fixture_identity_and_stability_are_required(self) -> None:
|
|
receipt = receipt_fixture()
|
|
budget = mod.budget_from_receipt(receipt)
|
|
receipt["representative_context"]["fixture_id"] = "ambient-developer-state"
|
|
with self.assertRaisesRegex(mod.RuntimeContractError, "fixture_id"):
|
|
mod.compare(receipt, budget)
|
|
|
|
receipt = receipt_fixture()
|
|
receipt["representative_context"]["prompts_byte_identical"] = False
|
|
with self.assertRaisesRegex(mod.RuntimeContractError, "must be true"):
|
|
mod.compare(receipt, budget)
|
|
|
|
def test_tool_identity_rejects_equal_size_substitution_and_removal(self) -> None:
|
|
budget = mod.budget_from_receipt(receipt_fixture())
|
|
|
|
receipt = receipt_fixture()
|
|
active = receipt["tool_catalog"]["modes"]["act"]["active"]
|
|
active["tool_names"] = sorted(["File", "Hash"])
|
|
active["identity_sha256"] = mod.tool_identity_digest(active["tool_names"])
|
|
with self.assertRaisesRegex(mod.RuntimeContractError, "identity changed"):
|
|
mod.compare(receipt, budget)
|
|
|
|
receipt = receipt_fixture()
|
|
full = receipt["tool_catalog"]["modes"]["plan"]["full"]
|
|
full["tool_names"].remove("Git")
|
|
full["tools"] -= 1
|
|
full["identity_sha256"] = mod.tool_identity_digest(full["tool_names"])
|
|
with self.assertRaisesRegex(mod.RuntimeContractError, "identity changed"):
|
|
mod.compare(receipt, budget)
|
|
|
|
def test_tool_identity_rejects_missing_structure_and_wrong_mode_swap(self) -> None:
|
|
budget = mod.budget_from_receipt(receipt_fixture())
|
|
|
|
receipt = receipt_fixture()
|
|
del receipt["tool_catalog"]["modes"]["operate"]["full"]["tool_names"]
|
|
with self.assertRaisesRegex(mod.RuntimeContractError, "missing required field"):
|
|
mod.compare(receipt, budget)
|
|
|
|
receipt = receipt_fixture()
|
|
plan = receipt["tool_catalog"]["modes"]["plan"]["active"]
|
|
act = receipt["tool_catalog"]["modes"]["act"]["active"]
|
|
for field in ["tool_names", "identity_sha256"]:
|
|
plan[field], act[field] = act[field], plan[field]
|
|
with self.assertRaisesRegex(mod.RuntimeContractError, "identity changed"):
|
|
mod.compare(receipt, budget)
|
|
|
|
def test_representative_stage_identity_rejects_same_length_substitution(self) -> None:
|
|
receipt = receipt_fixture()
|
|
budget = mod.budget_from_receipt(receipt)
|
|
digest = receipt["representative_context"]["stages"]["project"][
|
|
"identity_sha256"
|
|
]
|
|
receipt["representative_context"]["stages"]["project"][
|
|
"identity_sha256"
|
|
] = ("0" if digest[0] != "0" else "1") + digest[1:]
|
|
with self.assertRaisesRegex(mod.RuntimeContractError, "identity changed"):
|
|
mod.compare(receipt, budget)
|
|
|
|
def test_same_resolved_receipt_and_budget_path_is_rejected(self) -> None:
|
|
budget = mod.budget_from_receipt(receipt_fixture())
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
budget_path = Path(tmp) / "budget.json"
|
|
nested = Path(tmp) / "nested"
|
|
nested.mkdir()
|
|
budget_path.write_text(json.dumps(budget), encoding="utf-8")
|
|
alias = nested / ".." / "budget.json"
|
|
errors = io.StringIO()
|
|
with redirect_stderr(errors):
|
|
result = mod.main(
|
|
["--receipt", str(alias), "--budget", str(budget_path)]
|
|
)
|
|
self.assertEqual(result, 2)
|
|
self.assertIn("distinct filesystem paths", errors.getvalue())
|
|
|
|
def test_receipt_path_avoids_running_measurement(self) -> None:
|
|
receipt = receipt_fixture()
|
|
budget = mod.budget_from_receipt(receipt)
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
receipt_path, budget_path = write_documents(tmp, receipt, budget)
|
|
output = io.StringIO()
|
|
with (
|
|
mock.patch.object(
|
|
mod, "run_measurement", side_effect=AssertionError("must stay hermetic")
|
|
),
|
|
redirect_stdout(output),
|
|
):
|
|
result = mod.main(
|
|
["--receipt", str(receipt_path), "--budget", str(budget_path)]
|
|
)
|
|
self.assertEqual(result, 0)
|
|
self.assertIn("all 55 metrics are exactly at budget", output.getvalue())
|
|
|
|
def test_default_measurement_forces_cargo_offline(self) -> None:
|
|
receipt = receipt_fixture()
|
|
completed = mock.Mock(returncode=0, stdout=json.dumps(receipt), stderr="")
|
|
with mock.patch.object(mod.subprocess, "run", return_value=completed) as run:
|
|
measured = mod.run_measurement()
|
|
self.assertEqual(measured, receipt)
|
|
command = run.call_args.args[0]
|
|
environment = run.call_args.kwargs["env"]
|
|
self.assertEqual(command, [sys.executable, str(mod.MEASURE_SCRIPT)])
|
|
self.assertEqual(environment["CARGO_NET_OFFLINE"], "true")
|
|
|
|
def test_cli_decrease_prints_a_tightening_command(self) -> None:
|
|
receipt = receipt_fixture()
|
|
budget = mod.budget_from_receipt(receipt)
|
|
set_path(receipt, ("system_prompt", "modes", "plan", "system_prompt_bytes"), 23000)
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
receipt_path, budget_path = write_documents(tmp, receipt, budget)
|
|
output = io.StringIO()
|
|
with redirect_stdout(output):
|
|
result = mod.main(
|
|
["--receipt", str(receipt_path), "--budget", str(budget_path)]
|
|
)
|
|
self.assertEqual(result, 0)
|
|
self.assertIn("1 can be tightened", output.getvalue())
|
|
self.assertIn("--update", output.getvalue())
|
|
|
|
def test_update_tightens_decreases_and_preserves_permissions(self) -> None:
|
|
receipt = receipt_fixture()
|
|
budget = mod.budget_from_receipt(receipt)
|
|
path = ("tool_catalog", "modes", "plan", "active", "bytes")
|
|
set_path(receipt, path, 11000)
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
receipt_path, budget_path = write_documents(tmp, receipt, budget)
|
|
os.chmod(budget_path, 0o640)
|
|
with redirect_stdout(io.StringIO()):
|
|
result = mod.main(
|
|
[
|
|
"--receipt",
|
|
str(receipt_path),
|
|
"--budget",
|
|
str(budget_path),
|
|
"--update",
|
|
]
|
|
)
|
|
updated = json.loads(budget_path.read_text(encoding="utf-8"))
|
|
updated_mode = stat.S_IMODE(budget_path.stat().st_mode)
|
|
self.assertEqual(result, 0)
|
|
self.assertEqual(mod.metric_value(updated, path, "budget"), 11000)
|
|
self.assertEqual(updated_mode, 0o640)
|
|
|
|
def test_atomic_update_failure_leaves_original_and_removes_temporary_file(self) -> None:
|
|
budget = mod.budget_from_receipt(receipt_fixture())
|
|
replacement = mod.budget_from_receipt(receipt_fixture())
|
|
replacement["_comment"] = "replacement"
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
budget_path = Path(tmp) / "budget.json"
|
|
original = json.dumps(budget)
|
|
budget_path.write_text(original, encoding="utf-8")
|
|
with (
|
|
mock.patch.object(mod.os, "replace", side_effect=OSError("stop")),
|
|
self.assertRaisesRegex(OSError, "stop"),
|
|
):
|
|
mod.write_budget_atomic(budget_path, replacement)
|
|
after = budget_path.read_text(encoding="utf-8")
|
|
temporary_files = list(Path(tmp).glob(".budget.json.*.tmp"))
|
|
self.assertEqual(after, original)
|
|
self.assertEqual(temporary_files, [])
|
|
|
|
def test_update_refuses_an_increase_without_rewriting_budget(self) -> None:
|
|
receipt = receipt_fixture()
|
|
budget = mod.budget_from_receipt(receipt)
|
|
path = ("skill_discovery", "second_delta", "directories_visited")
|
|
set_path(receipt, path, 2)
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
receipt_path, budget_path = write_documents(tmp, receipt, budget)
|
|
original = budget_path.read_text(encoding="utf-8")
|
|
with redirect_stderr(io.StringIO()):
|
|
result = mod.main(
|
|
[
|
|
"--receipt",
|
|
str(receipt_path),
|
|
"--budget",
|
|
str(budget_path),
|
|
"--update",
|
|
]
|
|
)
|
|
after = budget_path.read_text(encoding="utf-8")
|
|
self.assertEqual(result, 1)
|
|
self.assertEqual(after, original)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(unittest.main())
|