"""The crib and the skill body are checked against the code they describe.
Both are prompt text: the crib sits in the head of every request, the body is where a cell goes for
the rest, and neither gets re-read by a reviewer once it works. These checks make drift fail a
build instead of reaching a model. They read the code for the truth: parameters, defaults, order,
shapes and fields all come from the objects the prompt is talking about.
"""
from __future__ import annotations
import inspect
import re
import unittest
from klbr import workspace as tree
from peer import ROOT
CONTRACT = ROOT / "klbr-core/src/runtime_turn/runtime-contract.md"
SKILL_BODY = ROOT / "skills/workspace/SKILL.md"
SECTION = "Source work goes through the tree verbs"
SHAPES = (
tree.Entry,
tree.File,
tree.Listing,
tree.SearchResult,
tree.TextMatch,
tree.RunResult,
)
PARAMETER = re.compile(r"^(?P<name>[a-z_][a-z0-9_]*)(?P<optional>\?)?(?:=(?P<default>.+))?$")
SHAPE_CALL = re.compile(r"^\s*(?P<shape>Entry|Listing|SearchResult|TextMatch|RunResult)\((?P<body>.*)\)")
def fenced(text: str) -> list[str]:
"""Every fenced python block, as source."""
blocks, cursor = [], 0
while True:
start = text.find("```python", cursor)
if start < 0:
return blocks
end = text.index("```", start + 9)
blocks.append(text[start + 9 : end])
cursor = end + 3
def crib_lines() -> list[str]:
start = CONTRACT.read_text().index(SECTION)
fenced_at = CONTRACT.read_text().index("```python", start)
block = CONTRACT.read_text()[fenced_at : CONTRACT.read_text().index("```", fenced_at + 9)]
return [line for line in block.splitlines()[1:] if line and not line.startswith("#")]
BINDING = re.compile(r"^\s*(?P<name>[A-Za-z_][A-Za-z0-9_]*)\s*=.*await\s+(?:tree\.)?")
def claimed_fields(source: str, *, bare: bool = False) -> set[str]:
"""Field names a block claims exist.
Two kinds are read as a claim: a field access on something one of our calls returned, and the
fields spelled out inside a shape's own parentheses (``Entry(path, relative, ...)``). Everything
else - ``str.splitlines()``, a glob like ``"*.rs"`` - is somebody else's vocabulary and is not
judged here. The receiver rule is what keeps this honest without pretending to be a type
checker: ``note`` bound from ``await read(...)`` is ours, ``text`` bound from ``note.text()``
is a string and its methods are the standard library's business.
"""
claimed: set[str] = set()
ours: set[str] = set()
lines = source.splitlines()
for line in lines:
stripped = re.sub(r'"[^"]*"', "", line)
binding = BINDING.match(stripped)
if binding and re.search(r"await\s+(?:open_file|ls|search|run|delete)\(", stripped):
ours.add(binding.group("name"))
spelled = SHAPE_CALL.match(stripped)
if spelled:
for part in spelled.group("body").split(","):
part = part.strip()
if re.fullmatch(r"[a-z_][a-z0-9_]*", part):
claimed.add(part)
for line in lines:
stripped = re.sub(r'"[^"]*"', "", line)
if bare:
# The crib states fields without binding anything first: everything after the return
# marker is a claim, and the block is short enough that there is nowhere to hide.
tail = stripped.partition("->")[2]
claimed |= set(re.findall(r"\.([a-z_][a-z0-9_]*)", tail))
for receiver, field in re.findall(r"\b([A-Za-z_][A-Za-z0-9_]*)\.([a-z_][a-z0-9_]*)", stripped):
if receiver in ours or receiver in {"e", "entry", "entries", "listing", "hits", "match", "matches"}:
claimed.add(field)
return claimed
def declared(name: str) -> list[tuple[str, bool, str]]:
"""The callable's parameters as (name, has default, rendered default)."""
parameters = []
for parameter in inspect.signature(getattr(tree, name)).parameters.values():
default = (
""
if parameter.default is parameter.empty
else "None" if parameter.default is None else repr(parameter.default).replace("'", '"')
)
parameters.append((parameter.name, parameter.default is not parameter.empty, default))
return parameters
def known_names() -> set[str]:
"""Everything the code defines for these objects: fields, parameters and the shapes."""
names = {name for shape in SHAPES for name in dir(shape)}
for verb in tree.PRELUDE_NAMES:
names |= set(inspect.signature(getattr(tree, verb)).parameters)
for shape in SHAPES:
for member in vars(shape).values():
if callable(member):
names |= set(inspect.signature(member).parameters)
return names
class ContractCribTests(unittest.TestCase):
def test_every_claimed_field_names_the_shape_that_owns_it(self):
"""A field after `->` belongs to the returned shape, or to the shape that prefixes it.
Fusing two shapes into one field list reads as if the container had the element's fields:
`SearchResult; .truncated, .relative .line .text` invites `result.relative`. Naming the
owner per clause is what keeps that from being written, so the owner is what is checked.
"""
owners = {shape.__name__: {field for field in dir(shape) if not field.startswith("_")} for shape in SHAPES}
clauses = []
for line in crib_lines():
tail = line.partition("->")[2]
if not tail:
continue
owner = re.match(r"\s*([A-Za-z_][A-Za-z0-9_]*)", tail).group(1)
for clause in tail.split(";"):
named = re.match(r"\s*([A-Za-z_][A-Za-z0-9_]*)", clause)
if named and named.group(1) in owners:
owner = named.group(1)
clauses.append((line.strip(), owner, re.findall(r"\.([a-z_][a-z0-9_]*)", clause)))
for line, owner, fields in clauses:
for field in fields:
self.assertIn(field, owners[owner], f"{field!r} is not {owner}'s in: {line}")
def test_every_named_parameter_is_real_in_declared_order(self):
for line in crib_lines():
# A signature line is a call, awaited or not; anything else is a caller's expression,
# checked by the test below.
call = line.strip()
if not re.match(r"^[a-z_][a-z0-9_]*\(", call):
continue
# Parameters are read only up to the return marker: what follows it is prose about
# what comes back, not part of the call.
text = call.partition("->")[0].partition("#")[0]
name = text.split("(", 1)[0].strip()
self.assertIn(name, tree.PRELUDE_NAMES, f"crib names {name!r}, which is not bound")
inside = text[text.index("(") + 1 : text.rindex(")")]
parameters = declared(name)
cursor = 0
for part in [piece.strip() for piece in inside.split(",")]:
match = PARAMETER.match(part)
self.assertIsNotNone(match, f"{name}: {part!r} is not a parameter")
found = match.group("name")
optional = match.group("optional") == "?"
default = (match.group("default") or "").strip()
remaining = parameters[cursor:]
index = next((i for i, (n, _, _) in enumerate(remaining) if n == found), None)
self.assertIsNotNone(index, f"{name} has no parameter {found!r} at this position")
_, has_default, real_default = remaining[index]
self.assertEqual(
optional,
has_default,
f"{name}({found}): crib says {'optional' if optional else 'required'}",
)
if optional:
self.assertEqual(default, real_default, f"{name}({found}) default")
cursor += index + 1
def test_expressions_in_the_crib_use_our_verbs_and_fields(self):
# A crib line that is not a signature is an expression a cell is meant to copy, so what it
# calls and what it reads off the result have to exist.
for line in crib_lines():
stripped = line.strip()
if re.match(r"^[a-z_][a-z0-9_]*\(", stripped) or not stripped:
continue
verbs = set(re.findall(r"await\s+([a-z_][a-z0-9_]*)\(", stripped))
self.assertTrue(verbs, f"crib expression calls nothing: {stripped!r}")
self.assertEqual(set(), verbs - set(tree.PRELUDE_NAMES), f"unknown verb in {stripped!r}")
fields = set(re.findall(r"\.([a-z_][a-z0-9_]*)\(", stripped))
self.assertEqual(
set(),
fields - known_names(),
f"expression calls a field that does not exist: {stripped!r}",
)
def test_the_crib_names_real_fields(self):
claimed = claimed_fields("\n".join(crib_lines()), bare=True)
self.assertTrue(claimed, "the crib stopped saying what a call returns")
self.assertEqual(set(), claimed - known_names(), "the crib names fields that do not exist")
def test_the_skill_body_documents_every_bound_verb(self):
# The body is where a cell goes for the rest, so a verb renamed or dropped in code cannot
# stay described there as if it existed.
body = SKILL_BODY.read_text()
missing = [name for name in tree.PRELUDE_NAMES if f"{name}(" not in body]
self.assertEqual([], missing, f"skills/workspace does not document {missing}")
def test_the_skill_body_names_real_fields(self):
claimed = claimed_fields("\n".join(fenced(SKILL_BODY.read_text())))
self.assertTrue(claimed, "the body stopped saying what a call returns")
self.assertEqual(set(), claimed - known_names(), "the body names fields that do not exist")