Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions .verify-floor.json

Large diffs are not rendered by default.

5 changes: 0 additions & 5 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -124,8 +124,3 @@ mypy_path = ["src", "tests"]
# It fails toward motion, not silence: verify.py prints the remaining count every run, and
# `.verify-floor.json`'s `mypy_exempt_max` FAILS if the list grows — so new untyped code in a clean
# module is now a red, and a module can never quietly rejoin the exempt set.
[[tool.mypy.overrides]]
module = [
"capability_advisor",
]
ignore_errors = true
29 changes: 14 additions & 15 deletions src/capability_advisor.py
Original file line number Diff line number Diff line change
Expand Up @@ -644,17 +644,17 @@ def advise(
if bound:
present = {m["capability_id"] for m in matched}
for cap_id, reason in bound.items():
cap = caps.get(cap_id)
if cap is None or cap.get("status") in {"retired", "superseded"}:
bound_cap: dict[str, Any] | None = caps.get(cap_id)
if bound_cap is None or bound_cap.get("status") in {"retired", "superseded"}:
continue
if cap_id not in present:
matched.append(
{
"capability_id": cap_id,
"matched_task_type": None,
"entrypoint": cap.get("entrypoint"),
"entrypoint": bound_cap.get("entrypoint"),
"bound_only": True,
**_usability(cap),
**_usability(bound_cap),
}
)
unmatched.pop(cap_id, None)
Expand Down Expand Up @@ -771,10 +771,9 @@ def should_reask(previous: dict | None, current_context: dict) -> dict:

# The work reclassified — e.g. an "implement" task that has moved on to writing tests. This is
# the highest-value trigger: it is exactly when a different capability becomes relevant.
now_types = set(
classify_task(str(current_context.get("task") or ""))
and [c["task_type"] for c in classify_task(str(current_context.get("task") or ""))]
)
now_types: set[str] = {
c["task_type"] for c in classify_task(str(current_context.get("task") or ""))
}
was_types = set(previous.get("task_types") or [])
if now_types and now_types != was_types:
reasons.append(f"task_reclassified:{','.join(sorted(now_types - was_types)) or 'narrowed'}")
Expand Down Expand Up @@ -1549,7 +1548,7 @@ def consult_text(surface: str, day: str) -> str:
def consult_phases(
*,
day: str | None = None,
surfaces: list[str] | None = None,
surfaces: list[Any] | None = None,
record: bool = True,
path=None,
) -> dict:
Expand Down Expand Up @@ -1862,7 +1861,7 @@ def transferable_concept(capability_id: str) -> str | None:
return declared or None


def consult_target(repository: str) -> str:
def consult_target(repository: str | None) -> str:
"""What this consult is about: `self`, `audited_repo`, or `unknown` when no repo was named."""
repo = str(repository or "").strip()
if not repo:
Expand Down Expand Up @@ -2013,7 +2012,7 @@ def evaluate_precondition(
declared = applies_to(capability_id)
needs = required_repo_fact(capability_id)
target = consult_target(repository)
out: list[Any] = {
out: dict[str, Any] = {
"applies_to": declared,
"scope_target": target,
"scope_match": None,
Expand Down Expand Up @@ -3854,9 +3853,9 @@ def _selftest_findability() -> None:
with tempfile.TemporaryDirectory(prefix="adv-find-") as td:
ledger = Path(td) / "capabilities.json"
capabilities.save({}, ledger)
site = Path(td) / "fake-skill.md"
site.write_text('consult with surface: "t-find:asked"\n')
CONSULT_SITES["t-find:asked"] = {"caller": str(site), "how": "synthetic"}
site_path = Path(td) / "fake-skill.md"
site_path.write_text('consult with surface: "t-find:asked"\n')
CONSULT_SITES["t-find:asked"] = {"caller": str(site_path), "how": "synthetic"}

inv = surfaces_binding(
["wide-cap", "asked-cap", "silent-cap", "absent-cap"], path=ledger
Expand All @@ -3879,7 +3878,7 @@ def _selftest_findability() -> None:
# PRESENT BUT NO LONGER NAMING ITS SURFACE IS DRIFT, and drift must LEAVE `reached`.
# Broken first by treating any readable file as verification, which let a renamed
# surface keep counting as consulted forever.
site.write_text("this file no longer mentions the surface at all\n")
site_path.write_text("this file no longer mentions the surface at all\n")
drifted = consulting_surfaces()
assert "t-find:asked" not in drifted["reached"], drifted["reached"][:8]
assert any(d["surface"] == "t-find:asked" for d in drifted["drifted"]), drifted
Expand Down
29 changes: 28 additions & 1 deletion src/repo_knowledge.py
Original file line number Diff line number Diff line change
Expand Up @@ -2523,9 +2523,36 @@ def fake_gh_failure(command, **_kwargs):
assert clustered[1]["cluster_key"] == "always_run_pytest_before_merging", clustered
finally:
p.unlink(missing_ok=True)

# ---- MALFORMED current_refs IS REJECTED, WHATEVER SHAPE THE MALFORMATION TAKES.
# `validate_current_refs` declares `refs: list[dict]`, and its caller in capability_compiler
# used to pass `cast(dict, contract.get("current_refs") or {})` -- a cast that named the wrong
# type over a fallback of the wrong type. Harmless in practice, because a dict and an empty
# list both fail `not isinstance(refs, list) or not refs` and take the same branch, which is
# exactly why nothing noticed: the annotation was wrong and the behaviour was right. That
# coincidence is the reason to pin it rather than trust it -- the cast is now `list`/`[]`, and
# if anyone "simplifies" the isinstance guard away, a dict would start being ITERATED (yielding
# its string keys) instead of rejected.
import tempfile as _tf

with _tf.TemporaryDirectory(prefix="rk-refs-") as _td:
# Annotated because the ratchet now checks this module: a bare heterogeneous tuple gives
# mypy nothing to infer. The shapes are deliberately mixed -- that IS the test.
_malformed: tuple[Any, ...] = ({}, {"path": "README.md"}, [], "README.md", None)
for _bad in _malformed:
_res = validate_current_refs(_td, _bad) # type: ignore[arg-type] # malformed on purpose
assert _res["valid"] is False, (_bad, _res)
assert _res["errors"] == ["current refs are required"], (_bad, _res)
assert _res["refs"] == [], (_bad, _res)
# and the well-formed shape still validates, so the assertions above are not vacuous
(Path(_td) / "README.md").write_text("x\n")
_ok = validate_current_refs(_td, [{"path": "README.md"}])
assert _ok["valid"] is True, _ok

print(
"repo_knowledge.py selftest: OK (seed, filters, prompt append, truncation, "
"snapshot/docs/review suggestions, memory search, clustering, approval)"
"snapshot/docs/review suggestions, memory search, clustering, approval, "
"malformed current_refs rejected)"
)


Expand Down
31 changes: 25 additions & 6 deletions src/verify.py
Original file line number Diff line number Diff line change
Expand Up @@ -532,9 +532,19 @@ def _appended_note(prior: str | None, collected: int, passed: int, today: str) -
def mypy_exempt_modules() -> list[str] | None:
"""Modules on `[[tool.mypy.overrides]] ignore_errors` — the ratchet's blocking quantity.

None means the question could not be answered here (no pyproject.toml, unreadable, no override).
REPORTED, never treated as zero: a ratchet that stops being counted is indistinguishable from
one that emptied, and only one of those is good news.
None means the question COULD NOT BE ANSWERED (no pyproject.toml, or it does not parse).
[] means it was answered and nothing is exempt -- the drained state. REPORTED, never treated as
zero: a ratchet that stops being counted is indistinguishable from one that emptied, and only
one of those is good news.

THOSE TWO WERE THE SAME VALUE UNTIL 2026-08-24, and the docstring above was already the argument
against it. A readable pyproject with no `ignore_errors` override returned None, so the run that
FINISHED the drain printed "NOT COUNTED" -- the gate going quiet at the exact moment it
succeeded, which is the failure this file exists to prevent. Two things prove it was unintended:
`_format_mypy_exempt_line` already carries a " -- fully drained" branch that no input could
reach, and the selftest below asserted this function was TRUTHY, so an empty list would have
failed it. A gate whose own test forbids its drained state is a latched gate (CLAUDE.md), and
this one would have latched on the last module.
"""
try:
import tomllib
Expand All @@ -546,7 +556,8 @@ def mypy_exempt_modules() -> list[str] | None:
if override.get("ignore_errors"):
mods = override.get("module")
return sorted(mods) if isinstance(mods, list) else [str(mods)]
return None
# Readable, and nothing is exempt: ANSWERED, not unanswerable. See the docstring.
return []


def _format_mypy_exempt_line(mods: list[str] | None, limit: int | None, total: int = 99) -> str:
Expand Down Expand Up @@ -1174,8 +1185,16 @@ def _selftest() -> None:
assert _ceiling_problems({"mypy_exempt_max": 64}, {**_zero, "mypy_exempt_max": 64}) == []
_over = _ceiling_problems({"mypy_exempt_max": 64}, {**_zero, "mypy_exempt_max": 65})
assert len(_over) == 1 and "mypy_exempt_max" in _over[0], _over
# And the real file must be readable, or the line would silently report NOT COUNTED forever.
assert mypy_exempt_modules(), "pyproject.toml's ignore_errors override is unreadable"
# The three states are DISTINCT, and conflating the last two is what made this a latched gate:
# a full drain must read as "fully drained", never as "NOT COUNTED".
assert "NOT COUNTED" in _format_mypy_exempt_line(None, 0), "unanswerable must say so"
assert "fully drained" in _format_mypy_exempt_line([], 0), "an empty ratchet is the good news"
assert "1/1 max" in _format_mypy_exempt_line(["m"], 1), "a live ratchet reports both numbers"
# And the real file must PARSE -- `is not None`, not truthiness. Asserting truthiness here meant
# the ratchet's own selftest failed the moment the list emptied, i.e. the gate forbade its drain.
assert (
mypy_exempt_modules() is not None
), "pyproject.toml does not parse; the ratchet cannot be counted"

print(
"verify.py selftest: OK (count parsing, selftest discovery, silent-zero-exit is a "
Expand Down
Loading