CodeWhale Bot
cae5626e6c
- Every session entry carries id + parentId, leafId tracks active position, in-memory tree projects from journal, context rebuilds root->leaf. - Tree operations as commands: /tree (render), /branch (move leaf only, never rewrites history), /fork (new session from any node, interactive picker per #576 via /fork picker), /resume (picker + foreign-session import/export container). - branch_summary and compaction entries are first-class SessionEntryKind variants (data shape lands now, strategies deferred). - Spawn-depth tracking on SessionMetadata and Journal; fork increments. - Foreign-session import/export via SessionImportContainer so /resume can ingest sessions from other agents. - SavedSession journal migration: old linear messages -> journal entries with linked parent chain, leaf = last; new sessions write both journal and derived messages for compat. Atomic write/fsync/crash-checkpoint and MAX_SESSIONS=50 preserved. - Shares entry shape with compaction (same SessionEntry envelope). Co-depends on #5261 engine split (core journal placeholder already landed in parallel work on same branch).
697 lines
26 KiB
Python
Executable File
697 lines
26 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Enforce one-way ceilings on Codewhale's provider-free runtime contract.
|
|
|
|
The default invocation runs ``measure-runtime-contract.py`` with Cargo forced
|
|
offline. Pass ``--receipt`` to check an existing JSON receipt without compiling,
|
|
which also keeps this checker's unit tests hermetic.
|
|
|
|
Usage:
|
|
python3 scripts/check-runtime-contract-budget.py
|
|
python3 scripts/check-runtime-contract-budget.py --receipt receipt.json
|
|
python3 scripts/check-runtime-contract-budget.py --update
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import shlex
|
|
import stat
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
from typing import Any, Sequence
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parent.parent
|
|
BUDGET_PATH = REPO_ROOT / "scripts" / "runtime-contract-budget.json"
|
|
MEASURE_SCRIPT = REPO_ROOT / "scripts" / "measure-runtime-contract.py"
|
|
RECEIPT_KIND = "codewhale.runtime_contract_receipt"
|
|
BUDGET_KIND = "codewhale.runtime_contract_budget"
|
|
SCHEMA_VERSION = 1
|
|
REPRESENTATIVE_FIXTURE_ID = "representative-v1"
|
|
TOOL_SURFACE_PROFILE = "production-default-builtins-no-mcp-no-host-interpreters-v1"
|
|
|
|
MetricPath = tuple[str, ...]
|
|
MetricResult = tuple[str, str, int, int]
|
|
|
|
VISIBLE_MODES = (("plan", "Plan"), ("act", "Act"), ("operate", "Operate"))
|
|
REPRESENTATIVE_STAGES = (
|
|
("base", "base"),
|
|
("project", "project-authority"),
|
|
("instructions", "configured-instructions"),
|
|
("skill", "skill"),
|
|
("memory", "memory"),
|
|
("goal", "goal"),
|
|
("handoff", "handoff"),
|
|
)
|
|
TOOL_SURFACES = (("full", "full"), ("active", "active"))
|
|
|
|
METRICS: tuple[tuple[MetricPath, str], ...] = (
|
|
*(
|
|
(("system_prompt", "modes", mode, field), f"{label} {description}")
|
|
for mode, label in VISIBLE_MODES
|
|
for field, description in (
|
|
("system_prompt_bytes", "system-prompt bytes"),
|
|
("system_prompt_tokens_est", "system-prompt estimated tokens"),
|
|
("system_prompt_blocks", "system-prompt blocks"),
|
|
("mode_instructions_bytes", "mode-instruction bytes"),
|
|
("mode_instructions_tokens_est", "mode-instruction estimated tokens"),
|
|
)
|
|
),
|
|
*(
|
|
(
|
|
("representative_context", "stages", stage, "bytes"),
|
|
f"representative {label} stage bytes",
|
|
)
|
|
for stage, label in REPRESENTATIVE_STAGES
|
|
),
|
|
*(
|
|
(
|
|
("representative_context", "stages", stage, "delta_bytes"),
|
|
f"representative {label} stage delta bytes",
|
|
)
|
|
for stage, label in REPRESENTATIVE_STAGES[1:]
|
|
),
|
|
(
|
|
("representative_context", "total_bytes"),
|
|
"representative total bytes",
|
|
),
|
|
(
|
|
("representative_context", "total_tokens_est"),
|
|
"representative estimated tokens",
|
|
),
|
|
(
|
|
("representative_context", "system_prompt_blocks"),
|
|
"representative system-prompt blocks",
|
|
),
|
|
*(
|
|
(
|
|
("tool_catalog", "modes", mode, surface, field),
|
|
f"{label} {surface_label} {description}",
|
|
)
|
|
for mode, label in VISIBLE_MODES
|
|
for surface, surface_label in TOOL_SURFACES
|
|
for field, description in (
|
|
("tools", "tool count"),
|
|
("bytes", "tool-schema bytes"),
|
|
("tokens_est", "tool-schema estimated tokens"),
|
|
)
|
|
),
|
|
(
|
|
("skill_discovery", "first_delta", "root_discovery_calls"),
|
|
"first unchanged-turn root discovery calls",
|
|
),
|
|
(
|
|
("skill_discovery", "first_delta", "directories_visited"),
|
|
"first unchanged-turn directories visited",
|
|
),
|
|
(
|
|
("skill_discovery", "first_delta", "skill_md_read_attempts"),
|
|
"first unchanged-turn SKILL.md read attempts",
|
|
),
|
|
(
|
|
("skill_discovery", "second_delta", "root_discovery_calls"),
|
|
"second unchanged-turn root discovery calls",
|
|
),
|
|
(
|
|
("skill_discovery", "second_delta", "directories_visited"),
|
|
"second unchanged-turn directories visited",
|
|
),
|
|
(
|
|
("skill_discovery", "second_delta", "skill_md_read_attempts"),
|
|
"second unchanged-turn SKILL.md read attempts",
|
|
),
|
|
)
|
|
|
|
IDENTITIES: tuple[tuple[MetricPath, str], ...] = (
|
|
(("tool_catalog", "surface_profile"), "tool surface profile"),
|
|
*(
|
|
(
|
|
("tool_catalog", "modes", mode, surface, field),
|
|
f"{label} {surface_label} tool {description}",
|
|
)
|
|
for mode, label in VISIBLE_MODES
|
|
for surface, surface_label in TOOL_SURFACES
|
|
for field, description in (
|
|
("tool_names", "names"),
|
|
("identity_sha256", "identity digest"),
|
|
)
|
|
),
|
|
*(
|
|
(
|
|
("representative_context", "stages", stage, "identity_sha256"),
|
|
f"representative {label} stage identity digest",
|
|
)
|
|
for stage, label in REPRESENTATIVE_STAGES
|
|
),
|
|
)
|
|
|
|
|
|
class RuntimeContractError(ValueError):
|
|
"""A receipt or budget is missing a required, well-typed metric."""
|
|
|
|
|
|
def load_json(path: Path, kind: str) -> dict[str, Any]:
|
|
try:
|
|
document = json.loads(path.read_text(encoding="utf-8"))
|
|
except FileNotFoundError as error:
|
|
raise RuntimeContractError(f"missing {kind}: {path}") from error
|
|
except (OSError, json.JSONDecodeError) as error:
|
|
raise RuntimeContractError(f"invalid {kind} {path}: {error}") from error
|
|
if not isinstance(document, dict):
|
|
raise RuntimeContractError(f"invalid {kind} {path}: top level must be an object")
|
|
return document
|
|
|
|
|
|
def validate_document(
|
|
document: dict[str, Any], expected_kind: str, source: str
|
|
) -> None:
|
|
actual_kind = document.get("document_kind")
|
|
if actual_kind != expected_kind:
|
|
raise RuntimeContractError(
|
|
f"{source} document_kind must be `{expected_kind}`, got {actual_kind!r}"
|
|
)
|
|
version = document.get("schema_version")
|
|
if (
|
|
isinstance(version, bool)
|
|
or not isinstance(version, int)
|
|
or version != SCHEMA_VERSION
|
|
):
|
|
raise RuntimeContractError(
|
|
f"{source} schema_version must be {SCHEMA_VERSION}, got {version!r}"
|
|
)
|
|
|
|
|
|
def required_value(document: dict[str, Any], path: MetricPath, kind: str) -> Any:
|
|
value: Any = document
|
|
dotted = ".".join(path)
|
|
for part in path:
|
|
if not isinstance(value, dict) or part not in value:
|
|
raise RuntimeContractError(f"{kind} is missing required field `{dotted}`")
|
|
value = value[part]
|
|
return value
|
|
|
|
|
|
def tool_identity_digest(names: list[str]) -> str:
|
|
return hashlib.sha256("\0".join(names).encode("utf-8")).hexdigest()
|
|
|
|
|
|
def validate_identity_structure(document: dict[str, Any], kind: str) -> None:
|
|
profile = required_value(document, ("tool_catalog", "surface_profile"), kind)
|
|
if profile != TOOL_SURFACE_PROFILE:
|
|
raise RuntimeContractError(
|
|
f"{kind} tool surface_profile must be `{TOOL_SURFACE_PROFILE}`, "
|
|
f"got {profile!r}"
|
|
)
|
|
|
|
for mode, _label in VISIBLE_MODES:
|
|
for surface, _surface_label in TOOL_SURFACES:
|
|
base = ("tool_catalog", "modes", mode, surface)
|
|
names = required_value(document, (*base, "tool_names"), kind)
|
|
dotted_names = ".".join((*base, "tool_names"))
|
|
if (
|
|
not isinstance(names, list)
|
|
or any(not isinstance(name, str) or not name for name in names)
|
|
or names != sorted(set(names))
|
|
):
|
|
raise RuntimeContractError(
|
|
f"{kind} field `{dotted_names}` must be sorted unique non-empty strings"
|
|
)
|
|
count = metric_value(document, (*base, "tools"), kind)
|
|
if count != len(names):
|
|
raise RuntimeContractError(
|
|
f"{kind} metric `{'.'.join((*base, 'tools'))}` must equal the "
|
|
f"owned tool_names length ({len(names)})"
|
|
)
|
|
digest = required_value(document, (*base, "identity_sha256"), kind)
|
|
expected = tool_identity_digest(names)
|
|
if digest != expected:
|
|
raise RuntimeContractError(
|
|
f"{kind} field `{'.'.join((*base, 'identity_sha256'))}` must "
|
|
"match the owned sorted tool_names"
|
|
)
|
|
|
|
for stage, _label in REPRESENTATIVE_STAGES:
|
|
path = ("representative_context", "stages", stage, "identity_sha256")
|
|
digest = required_value(document, path, kind)
|
|
if not isinstance(digest, str) or re.fullmatch(r"[0-9a-f]{64}", digest) is None:
|
|
raise RuntimeContractError(
|
|
f"{kind} field `{'.'.join(path)}` must be a lowercase SHA-256 digest"
|
|
)
|
|
|
|
|
|
def validate_receipt(receipt: dict[str, Any]) -> None:
|
|
validate_document(receipt, RECEIPT_KIND, "receipt")
|
|
skill_discovery = receipt.get("skill_discovery")
|
|
identical = (
|
|
skill_discovery.get("prompts_byte_identical")
|
|
if isinstance(skill_discovery, dict)
|
|
else None
|
|
)
|
|
if identical is not True:
|
|
raise RuntimeContractError(
|
|
"receipt metric `skill_discovery.prompts_byte_identical` must be true"
|
|
)
|
|
representative = receipt.get("representative_context")
|
|
fixture_id = (
|
|
representative.get("fixture_id")
|
|
if isinstance(representative, dict)
|
|
else None
|
|
)
|
|
if fixture_id != REPRESENTATIVE_FIXTURE_ID:
|
|
raise RuntimeContractError(
|
|
"receipt metric `representative_context.fixture_id` must be "
|
|
f"`{REPRESENTATIVE_FIXTURE_ID}`, got {fixture_id!r}"
|
|
)
|
|
representative_identical = representative.get("prompts_byte_identical")
|
|
if representative_identical is not True:
|
|
raise RuntimeContractError(
|
|
"receipt metric `representative_context.prompts_byte_identical` must be true"
|
|
)
|
|
validate_identity_structure(receipt, "receipt")
|
|
|
|
|
|
def validate_budget(budget: dict[str, Any]) -> None:
|
|
validate_document(budget, BUDGET_KIND, "budget")
|
|
representative = budget.get("representative_context")
|
|
fixture_id = (
|
|
representative.get("fixture_id")
|
|
if isinstance(representative, dict)
|
|
else None
|
|
)
|
|
if fixture_id != REPRESENTATIVE_FIXTURE_ID:
|
|
raise RuntimeContractError(
|
|
"budget metric `representative_context.fixture_id` must be "
|
|
f"`{REPRESENTATIVE_FIXTURE_ID}`, got {fixture_id!r}"
|
|
)
|
|
validate_identity_structure(budget, "budget")
|
|
|
|
|
|
def metric_value(document: dict[str, Any], path: MetricPath, kind: str) -> int:
|
|
value = required_value(document, path, kind)
|
|
dotted = ".".join(path)
|
|
if isinstance(value, bool) or not isinstance(value, int) or value < 0:
|
|
raise RuntimeContractError(
|
|
f"{kind} metric `{dotted}` must be a non-negative integer"
|
|
)
|
|
return value
|
|
|
|
|
|
def compare(
|
|
receipt: dict[str, Any], budget: dict[str, Any]
|
|
) -> tuple[list[MetricResult], list[MetricResult]]:
|
|
"""Return (increases, decreases) as path/label/current/ceiling tuples."""
|
|
validate_receipt(receipt)
|
|
validate_budget(budget)
|
|
for path, label in IDENTITIES:
|
|
receipt_value = required_value(receipt, path, "receipt")
|
|
budget_value = required_value(budget, path, "budget")
|
|
if receipt_value != budget_value:
|
|
detail = ""
|
|
if isinstance(receipt_value, list) and isinstance(budget_value, list):
|
|
added = [str(item) for item in receipt_value if item not in budget_value]
|
|
removed = [
|
|
str(item) for item in budget_value if item not in receipt_value
|
|
]
|
|
detail = f" (added={added} removed={removed})"
|
|
raise RuntimeContractError(
|
|
f"identity changed for {label} [`{'.'.join(path)}`]{detail}"
|
|
)
|
|
increases: list[MetricResult] = []
|
|
decreases: list[MetricResult] = []
|
|
for path, label in METRICS:
|
|
current = metric_value(receipt, path, "receipt")
|
|
ceiling = metric_value(budget, path, "budget")
|
|
result = (".".join(path), label, current, ceiling)
|
|
if current > ceiling:
|
|
increases.append(result)
|
|
elif current < ceiling:
|
|
decreases.append(result)
|
|
return increases, decreases
|
|
|
|
|
|
def set_path_value(document: dict[str, Any], path: MetricPath, value: Any) -> None:
|
|
target = document
|
|
for part in path[:-1]:
|
|
target = target.setdefault(part, {})
|
|
target[path[-1]] = value
|
|
|
|
|
|
def budget_from_receipt(receipt: dict[str, Any]) -> dict[str, Any]:
|
|
validate_receipt(receipt)
|
|
budget: dict[str, Any] = {
|
|
"_comment": (
|
|
"One-way numeric ceilings and exact structural identities for the "
|
|
"provider-free runtime contract. Decreases pass; increases or identity "
|
|
"changes fail. Lock in decreases with: python3 "
|
|
"scripts/check-runtime-contract-budget.py --update"
|
|
),
|
|
"document_kind": BUDGET_KIND,
|
|
"schema_version": SCHEMA_VERSION,
|
|
"representative_context": {
|
|
"fixture_id": REPRESENTATIVE_FIXTURE_ID,
|
|
},
|
|
}
|
|
for path, _label in METRICS:
|
|
set_path_value(budget, path, metric_value(receipt, path, "receipt"))
|
|
for path, _label in IDENTITIES:
|
|
set_path_value(
|
|
budget,
|
|
path,
|
|
copy.deepcopy(required_value(receipt, path, "receipt")),
|
|
)
|
|
return budget
|
|
|
|
|
|
def write_budget_atomic(path: Path, budget: dict[str, Any]) -> None:
|
|
"""Replace an existing budget atomically without changing its mode bits."""
|
|
original_mode = stat.S_IMODE(path.stat().st_mode)
|
|
payload = json.dumps(budget, indent=2, sort_keys=True) + "\n"
|
|
file_descriptor, temporary_name = tempfile.mkstemp(
|
|
prefix=f".{path.name}.", suffix=".tmp", dir=path.parent
|
|
)
|
|
temporary_path = Path(temporary_name)
|
|
try:
|
|
with os.fdopen(file_descriptor, "w", encoding="utf-8") as handle:
|
|
handle.write(payload)
|
|
handle.flush()
|
|
os.fsync(handle.fileno())
|
|
os.chmod(temporary_path, original_mode)
|
|
os.replace(temporary_path, path)
|
|
except BaseException:
|
|
temporary_path.unlink(missing_ok=True)
|
|
raise
|
|
|
|
|
|
def run_measurement() -> dict[str, Any]:
|
|
env = os.environ.copy()
|
|
env["CARGO_NET_OFFLINE"] = "true"
|
|
proc = subprocess.run(
|
|
[sys.executable, str(MEASURE_SCRIPT)],
|
|
cwd=REPO_ROOT,
|
|
env=env,
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
sys.stderr.write(proc.stderr)
|
|
if proc.returncode != 0:
|
|
sys.stdout.write(proc.stdout)
|
|
raise RuntimeContractError(
|
|
f"runtime-contract measurement failed with exit code {proc.returncode}"
|
|
)
|
|
try:
|
|
receipt = json.loads(proc.stdout)
|
|
except json.JSONDecodeError as error:
|
|
raise RuntimeContractError(f"measurement emitted invalid JSON: {error}") from error
|
|
if not isinstance(receipt, dict):
|
|
raise RuntimeContractError("measurement top level must be an object")
|
|
validate_receipt(receipt)
|
|
return receipt
|
|
|
|
|
|
def update_command(receipt_path: Path | None, budget_path: Path) -> str:
|
|
parts = ["python3", "scripts/check-runtime-contract-budget.py"]
|
|
if receipt_path is not None:
|
|
parts.extend(["--receipt", str(receipt_path)])
|
|
if budget_path != BUDGET_PATH:
|
|
parts.extend(["--budget", str(budget_path)])
|
|
parts.append("--update")
|
|
return shlex.join(parts)
|
|
|
|
|
|
FRAGMENT_MODULE = REPO_ROOT / "crates" / "core" / "src" / "fragments.rs"
|
|
FRAGMENT_MAX_TOKENS_CEILING = 10_000
|
|
FRAGMENT_MAX_BYTES_CEILING = FRAGMENT_MAX_TOKENS_CEILING * 4
|
|
FRAGMENT_DEFAULT_MAX_BYTES_CEILING = 4 * 1024
|
|
FRAGMENT_MAX_COUNT_CEILING = 16
|
|
|
|
|
|
def check_fragment_caps() -> None:
|
|
"""Gate the bounded fragment hard caps (issue #5264).
|
|
|
|
Static check — no cargo needed. Fails closed if the fragment module is
|
|
missing, if any cap has been raised without review, or if the
|
|
project-instruction import is absent.
|
|
"""
|
|
try:
|
|
text = FRAGMENT_MODULE.read_text(encoding="utf-8")
|
|
except FileNotFoundError as error:
|
|
raise RuntimeContractError(
|
|
f"missing bounded fragment module: {FRAGMENT_MODULE} ({error})"
|
|
) from error
|
|
|
|
def const_value(pattern: str) -> int:
|
|
match = re.search(pattern, text)
|
|
if not match:
|
|
raise RuntimeContractError(f"fragment cap missing: {pattern}")
|
|
try:
|
|
return int(match.group(1).replace("_", ""))
|
|
except ValueError as error:
|
|
raise RuntimeContractError(f"fragment cap not an int: {pattern}") from error
|
|
|
|
max_tokens = const_value(r"pub const MAX_FRAGMENT_TOKENS:\s*usize\s*=\s*([0-9_]+)")
|
|
if max_tokens != FRAGMENT_MAX_TOKENS_CEILING:
|
|
raise RuntimeContractError(
|
|
f"MAX_FRAGMENT_TOKENS must be {FRAGMENT_MAX_TOKENS_CEILING}, got {max_tokens}"
|
|
)
|
|
# MAX_FRAGMENT_BYTES must be defined as MAX_FRAGMENT_TOKENS * 4 (canonical)
|
|
# or as a literal 40000. Either way the derived ceiling is 40_000.
|
|
has_multiplication = re.search(
|
|
r"pub const MAX_FRAGMENT_BYTES:\s*usize\s*=\s*MAX_FRAGMENT_TOKENS\s*\*\s*4", text
|
|
)
|
|
bytes_literal = re.search(
|
|
r"pub const MAX_FRAGMENT_BYTES:\s*usize\s*=\s*([0-9_]+)", text
|
|
)
|
|
if bytes_literal:
|
|
literal = int(bytes_literal.group(1).replace("_", ""))
|
|
if literal != FRAGMENT_MAX_BYTES_CEILING:
|
|
raise RuntimeContractError(
|
|
f"MAX_FRAGMENT_BYTES must be {FRAGMENT_MAX_BYTES_CEILING}, got {literal}"
|
|
)
|
|
elif not has_multiplication:
|
|
raise RuntimeContractError(
|
|
"MAX_FRAGMENT_BYTES must be defined as MAX_FRAGMENT_TOKENS * 4 or as 40000"
|
|
)
|
|
# DEFAULT is defined as 4 * 1024 (canonical) or 4096 literal
|
|
has_default_multiplication = re.search(
|
|
r"pub const DEFAULT_FRAGMENT_MAX_BYTES:\s*usize\s*=\s*4\s*\*\s*1024", text
|
|
)
|
|
default_literal = re.search(
|
|
r"pub const DEFAULT_FRAGMENT_MAX_BYTES:\s*usize\s*=\s*([0-9_]+)", text
|
|
)
|
|
if has_default_multiplication:
|
|
# canonical 4*1024 == 4096, which equals ceiling
|
|
pass
|
|
elif default_literal:
|
|
default_bytes = int(default_literal.group(1).replace("_", ""))
|
|
if default_bytes != FRAGMENT_DEFAULT_MAX_BYTES_CEILING:
|
|
raise RuntimeContractError(
|
|
f"DEFAULT_FRAGMENT_MAX_BYTES must be {FRAGMENT_DEFAULT_MAX_BYTES_CEILING}, got {default_bytes}"
|
|
)
|
|
if default_bytes > FRAGMENT_MAX_BYTES_CEILING:
|
|
raise RuntimeContractError(
|
|
f"DEFAULT_FRAGMENT_MAX_BYTES ({default_bytes}) must not exceed MAX_FRAGMENT_BYTES ({FRAGMENT_MAX_BYTES_CEILING})"
|
|
)
|
|
else:
|
|
raise RuntimeContractError("DEFAULT_FRAGMENT_MAX_BYTES definition not found")
|
|
|
|
max_count = const_value(
|
|
r"pub const MAX_FRAGMENTS_PER_CONTEXT:\s*usize\s*=\s*([0-9_]+)"
|
|
)
|
|
if max_count != FRAGMENT_MAX_COUNT_CEILING:
|
|
raise RuntimeContractError(
|
|
f"MAX_FRAGMENTS_PER_CONTEXT must be {FRAGMENT_MAX_COUNT_CEILING}, got {max_count}"
|
|
)
|
|
if max_count > FRAGMENT_MAX_COUNT_CEILING:
|
|
raise RuntimeContractError(
|
|
f"MAX_FRAGMENTS_PER_CONTEXT ({max_count}) must not exceed {FRAGMENT_MAX_COUNT_CEILING}"
|
|
)
|
|
|
|
# Ensure every injection type is in FragmentId::all() and the
|
|
# project-instruction import is present as a typed fragment.
|
|
required_fragments = [
|
|
"Workspace",
|
|
"Permissions",
|
|
"Route",
|
|
"AgentTopology",
|
|
"SkillsTools",
|
|
"TokenBudget",
|
|
"ProjectInstructions",
|
|
"Constitution",
|
|
]
|
|
for name in required_fragments:
|
|
if f"Self::{name}" not in text and f"{name} =>" not in text and f'"{name.lower()}"' not in text.lower():
|
|
# Fallback: search for enum variant declaration
|
|
if not re.search(rf"\b{name}\b", text):
|
|
raise RuntimeContractError(
|
|
f"FragmentId missing required variant {name}"
|
|
)
|
|
# Marker stability — these strings are pinned by tests / prefix cache
|
|
required_markers = [
|
|
"<!-- cw:ctx:workspace -->",
|
|
"<!-- cw:ctx:project_instructions -->",
|
|
"<!-- cw:ctx:constitution -->",
|
|
]
|
|
for marker in required_markers:
|
|
if marker not in text:
|
|
raise RuntimeContractError(
|
|
f"bounded fragment module missing required marker {marker!r}"
|
|
)
|
|
|
|
# Project-instruction import must be a typed fragment, not ad-hoc
|
|
if "load_project_instruction_fragment" not in text:
|
|
raise RuntimeContractError(
|
|
"bounded fragment module must expose load_project_instruction_fragment (project-instruction import as typed fragment)"
|
|
)
|
|
if "PROJECT_INSTRUCTION_CANDIDATES" not in text:
|
|
raise RuntimeContractError(
|
|
"bounded fragment module must define PROJECT_INSTRUCTION_CANDIDATES"
|
|
)
|
|
# Required candidate files from #3978
|
|
required_candidates = [
|
|
".cursorrules",
|
|
".clinerules",
|
|
".windsurf/rules",
|
|
".gemini",
|
|
".github/copilot-instructions.md",
|
|
]
|
|
for candidate in required_candidates:
|
|
if candidate not in text:
|
|
raise RuntimeContractError(
|
|
f"PROJECT_INSTRUCTION_CANDIDATES missing required entry {candidate!r}"
|
|
)
|
|
|
|
# matches_text recognizer must exist on the fragment trait
|
|
if "fn matches_text" not in text:
|
|
raise RuntimeContractError(
|
|
"bounded fragment module must define a matches_text recognizer on the fragment trait"
|
|
)
|
|
if "trait ContextFragment" not in text:
|
|
raise RuntimeContractError(
|
|
"bounded fragment module must define trait ContextFragment with matches_text"
|
|
)
|
|
|
|
# No unbounded fragment — enforce that creation clamps to MAX_FRAGMENT_BYTES
|
|
if "MAX_FRAGMENT_BYTES" not in text or "enforce_byte_cap" not in text:
|
|
raise RuntimeContractError(
|
|
"bounded fragment module must enforce byte caps via enforce_byte_cap and MAX_FRAGMENT_BYTES"
|
|
)
|
|
|
|
# TUI must be unified with the core boundary (shared crates/core module)
|
|
tui_fragment = REPO_ROOT / "crates" / "tui" / "src" / "model_context" / "fragment.rs"
|
|
try:
|
|
tui_text = tui_fragment.read_text(encoding="utf-8")
|
|
except FileNotFoundError as error:
|
|
raise RuntimeContractError(
|
|
f"missing TUI fragment module: {tui_fragment} ({error})"
|
|
) from error
|
|
if "codewhale_core::fragments" not in tui_text:
|
|
raise RuntimeContractError(
|
|
"TUI model_context/fragment.rs must re-export caps from codewhale_core::fragments (shared crates/core boundary)"
|
|
)
|
|
if "ProjectInstructions" not in tui_text:
|
|
raise RuntimeContractError(
|
|
"TUI fragment module must include ProjectInstructions variant (unified with core)"
|
|
)
|
|
if "MAX_FRAGMENT_BYTES" not in tui_text:
|
|
raise RuntimeContractError(
|
|
"TUI fragment module must enforce MAX_FRAGMENT_BYTES (10K-token ceiling)"
|
|
)
|
|
if "matches_text" not in tui_text:
|
|
raise RuntimeContractError(
|
|
"TUI fragment module must expose a matches_text recognizer"
|
|
)
|
|
|
|
|
|
def main(argv: Sequence[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument(
|
|
"--receipt",
|
|
type=Path,
|
|
help="check an existing measurement JSON instead of compiling",
|
|
)
|
|
parser.add_argument(
|
|
"--budget",
|
|
type=Path,
|
|
default=BUDGET_PATH,
|
|
help=argparse.SUPPRESS,
|
|
)
|
|
parser.add_argument(
|
|
"--update",
|
|
action="store_true",
|
|
help="tighten all ceilings to the current receipt; refuses increases",
|
|
)
|
|
args = parser.parse_args(argv)
|
|
|
|
try:
|
|
check_fragment_caps()
|
|
if args.receipt is not None and args.receipt.resolve() == args.budget.resolve():
|
|
raise RuntimeContractError(
|
|
"receipt and budget must resolve to distinct filesystem paths"
|
|
)
|
|
budget = load_json(args.budget, "budget")
|
|
receipt = (
|
|
load_json(args.receipt, "receipt")
|
|
if args.receipt is not None
|
|
else run_measurement()
|
|
)
|
|
increases, decreases = compare(receipt, budget)
|
|
except RuntimeContractError as error:
|
|
print(f"[runtime-contract-budget] ERROR: {error}", file=sys.stderr)
|
|
return 2
|
|
|
|
if increases:
|
|
print("[runtime-contract-budget] FAIL: runtime contract grew:", file=sys.stderr)
|
|
for path, label, current, ceiling in increases:
|
|
print(
|
|
f" {label}: {current} > {ceiling} (+{current - ceiling}) [{path}]",
|
|
file=sys.stderr,
|
|
)
|
|
print(
|
|
"\nReduce the model-facing surface or make any higher ceiling an explicit "
|
|
"maintainer decision in scripts/runtime-contract-budget.json.",
|
|
file=sys.stderr,
|
|
)
|
|
return 1
|
|
|
|
if args.update:
|
|
try:
|
|
write_budget_atomic(args.budget, budget_from_receipt(receipt))
|
|
except OSError as error:
|
|
print(
|
|
f"[runtime-contract-budget] ERROR: failed to update budget: {error}",
|
|
file=sys.stderr,
|
|
)
|
|
return 2
|
|
print(
|
|
f"[runtime-contract-budget] wrote {args.budget}: "
|
|
f"{len(decreases)} decreased ceilings locked in "
|
|
f"({len(METRICS)} total)"
|
|
)
|
|
return 0
|
|
|
|
if decreases:
|
|
print(
|
|
f"[runtime-contract-budget] PASS: {len(METRICS)} ceilings respected; "
|
|
f"{len(decreases)} can be tightened."
|
|
)
|
|
for path, label, current, ceiling in decreases:
|
|
print(f" {label}: {current} < {ceiling} (-{ceiling - current}) [{path}]")
|
|
print(f"Tighten with:\n {update_command(args.receipt, args.budget)}")
|
|
return 0
|
|
|
|
print(
|
|
f"[runtime-contract-budget] PASS: all {len(METRICS)} metrics are exactly at budget."
|
|
)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|