russell@unturf.com
d8469613ce
The three "92 claim_pack docs" tags were drifting against shard 000.db's
21 rows because the harness only counted one shard, but the doc prose
("#000031 closed at 92") meant the corpus total (21+16+38+17 across
genesis shards 000-003).
Two-line fix path: extend the harness to sum across all ???.db shards
via a `*:` prefix (e.g. `*:documents?source_type=claim_pack`), then
prefix the three drifted tags. Aligns the harness scope with the
semantic scope of the claim instead of forcing the claim to shrink to
one shard.
The `*:` glob:
- Matches `[0-9][0-9][0-9].db` basenames only (operator sidecars
qa.db / snapshots.db / selfmodel-chain.db skipped)
- Skips shards lacking the named table (schema-version tolerance)
- Returns _DB_MISSING when no genesis shard exists (CI / fresh-
checkout skip semantic preserved)
- Returns _TABLE_MISSING when no contributing shard has the table
Documented in ticket-000044 §3.4 + a third example showing the new
syntax. Diagnosis credit to a sub-agent investigation that confirmed
zero eviction/falsification audit events on claim_packs — the data is
intact; the harness was just single-shard.
433 lines
15 KiB
Python
433 lines
15 KiB
Python
"""Regression test for AUTOCOUNT-tagged numeric claims in ``docs/``.
|
|
|
|
The doc-drift pattern recurred four times on 2026-05-10 (commits
|
|
``6cbbf95``, ``14bcb99``, ``5c21e83``, ``30a9488``). Test counts and
|
|
alias counts in docs went stale within hours of fox writing them
|
|
because more work landed in parallel. Each refresh cost a
|
|
Read + Edit + commit cycle.
|
|
|
|
This test makes drift loud at test time instead of catching it
|
|
visually. Tag any numeric claim in ``docs/**/*.md`` with the
|
|
HTML-comment pair below; the test compares the claimed number to
|
|
the live value and fails with a clear diff message.
|
|
|
|
Format::
|
|
|
|
<!--AUTOCOUNT:metric:path-->N<!--/AUTOCOUNT-->
|
|
|
|
Where ``metric`` is one of:
|
|
|
|
- ``tests`` — ``path`` is a pytest-collectible file/directory; the
|
|
live value is ``pytest --collect-only`` count.
|
|
- ``fixture-rows`` — ``path`` is a JSONL fixture; the live value
|
|
is the non-blank-non-comment line count.
|
|
- ``db-rows`` — ``path`` is either ``<table>`` (resolves against
|
|
the default shard ``~/.arborist/shards/000.db``) or
|
|
``<shard>:<table>`` (resolves against ``~/.arborist/shards/<shard>``).
|
|
When the DB or table isn't present (CI or fresh checkout), the
|
|
claim is skipped with a logged note rather than failing — claim-
|
|
pack and alias counts are operator state, not source-state.
|
|
- ``db-where`` — single-column equality predicate. ``path`` syntax::
|
|
|
|
<table>?<column>=<value>
|
|
<shard>:<table>?<column>=<value>
|
|
|
|
Resolves to ``SELECT COUNT(*) FROM <table> WHERE <column> = ?``
|
|
with ``<value>`` bound as a parameter (no SQL injection through
|
|
the value). Same skip-on-absence semantics as ``db-rows``. Use
|
|
for filtered-row claims like ``92 claim-pack records``::
|
|
|
|
<!--AUTOCOUNT:db-where:documents?source_type=claim_pack-->92<!--/AUTOCOUNT-->
|
|
|
|
GitHub and most markdown renderers strip HTML comments, so readers
|
|
see only ``N``. The tags are invisible in rendered output but
|
|
make the claim machine-checkable.
|
|
|
|
Adding a new claim: write the surrounding prose with the number,
|
|
wrap the number in the AUTOCOUNT comment pair, save, run
|
|
``pytest tests/test_doc_counts.py``. If it passes, ship.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
from collections import defaultdict
|
|
from pathlib import Path
|
|
from typing import Callable, Iterator
|
|
|
|
import pytest
|
|
|
|
REPO_ROOT = Path(__file__).parent.parent
|
|
DOCS_DIR = REPO_ROOT / "docs"
|
|
|
|
AUTOCOUNT_RE = re.compile(
|
|
r"<!--\s*AUTOCOUNT:(?P<metric>[a-z][a-z0-9_-]*)"
|
|
r":(?P<path>[^\s>]+)\s*-->"
|
|
r"(?P<n>\d+)"
|
|
r"<!--\s*/AUTOCOUNT\s*-->"
|
|
)
|
|
|
|
|
|
def _live_test_count_batch(paths: list[str]) -> dict[str, int]:
|
|
"""Run ``pytest --collect-only`` once for every requested path.
|
|
|
|
Returns a {path: count} map. We invoke pytest as a subprocess so
|
|
we don't recursively collect ourselves. Single batch keeps cost
|
|
near constant (~0.5s) regardless of how many docs reference
|
|
test files.
|
|
"""
|
|
if not paths:
|
|
return {}
|
|
out = subprocess.run(
|
|
[sys.executable, "-m", "pytest", "--collect-only", "-q", *paths],
|
|
capture_output=True,
|
|
text=True,
|
|
cwd=str(REPO_ROOT),
|
|
check=False,
|
|
)
|
|
counts: dict[str, int] = defaultdict(int)
|
|
for line in out.stdout.splitlines():
|
|
if "::" not in line:
|
|
continue
|
|
# Lines look like: tests/test_foo.py::test_bar
|
|
# or: tests/test_foo.py::TestClass::test_bar[case]
|
|
file_part = line.split("::", 1)[0].strip()
|
|
counts[file_part] += 1
|
|
# Make sure every requested path got a key (zero if collected nothing).
|
|
for p in paths:
|
|
counts.setdefault(p, 0)
|
|
return dict(counts)
|
|
|
|
|
|
def _live_fixture_rows(path: str) -> int:
|
|
full = REPO_ROOT / path
|
|
if not full.exists():
|
|
return -1
|
|
return sum(
|
|
1
|
|
for ln in full.read_text().splitlines()
|
|
if ln.strip() and not ln.strip().startswith("#")
|
|
)
|
|
|
|
|
|
_DB_MISSING = -2
|
|
_TABLE_MISSING = -3
|
|
_DB_ERROR = -4
|
|
|
|
# Default location for arborist alias / claim-pack tables. Operator state,
|
|
# not source state — absence is not a failure, it's a skip signal.
|
|
_DEFAULT_SHARDS_DIR = Path.home() / ".arborist" / "shards"
|
|
_DEFAULT_SHARD = _DEFAULT_SHARDS_DIR / "000.db"
|
|
|
|
|
|
def _live_db_rows(target: str) -> int:
|
|
"""Count rows in a SQLite table.
|
|
|
|
Target syntax::
|
|
|
|
<table> # ~/.arborist/shards/000.db (default)
|
|
<shard>:<table> # ~/.arborist/shards/<shard>
|
|
|
|
Sentinel returns:
|
|
_DB_MISSING (-2) DB file not present — caller marks skipped
|
|
_TABLE_MISSING (-3) DB present but table absent — caller marks skipped
|
|
_DB_ERROR (-4) sqlite3 error — caller marks skipped
|
|
N >= 0 live row count
|
|
"""
|
|
import sqlite3
|
|
|
|
if ":" in target:
|
|
shard_name, table = target.split(":", 1)
|
|
db = _DEFAULT_SHARDS_DIR / shard_name
|
|
else:
|
|
db = _DEFAULT_SHARD
|
|
table = target
|
|
if not db.exists():
|
|
return _DB_MISSING
|
|
# Validate table name is a bare identifier — defends against the
|
|
# dynamic SQL string interpolation below. Tags are author-controlled
|
|
# but this is belt-and-suspenders.
|
|
if not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", table):
|
|
return _DB_ERROR
|
|
try:
|
|
c = sqlite3.connect(f"file:{db}?mode=ro", uri=True)
|
|
try:
|
|
row = c.execute(
|
|
f"SELECT COUNT(*) FROM {table}" # noqa: S608 — validated above
|
|
).fetchone()
|
|
return int(row[0]) if row else 0
|
|
finally:
|
|
c.close()
|
|
except sqlite3.OperationalError as exc:
|
|
if "no such table" in str(exc).lower():
|
|
return _TABLE_MISSING
|
|
return _DB_ERROR
|
|
except sqlite3.Error:
|
|
return _DB_ERROR
|
|
|
|
|
|
def _live_db_where(target: str) -> int:
|
|
"""Count rows matching a single-column equality predicate.
|
|
|
|
Target syntax::
|
|
|
|
<table>?<column>=<value> # default shard 000.db
|
|
<shard>:<table>?<column>=<value> # one named shard
|
|
*:<table>?<column>=<value> # sum across ALL ???.db shards
|
|
|
|
The ``*`` form is for claims that span the corpus (e.g. "92
|
|
claim_pack docs" is 21+16+38+17 across genesis shards 000-003).
|
|
Globs match basenames ``[0-9][0-9][0-9].db`` only — operator
|
|
sidecar dbs (qa.db, snapshots.db) aren't summed.
|
|
|
|
Same sentinel returns as ``_live_db_rows``. ``<value>`` is bound
|
|
as a SQL parameter (no string interpolation), so even author-
|
|
typo'd or malicious values can't escape the predicate. Column
|
|
name is validated as a bare identifier (it goes into the SQL
|
|
text).
|
|
"""
|
|
import sqlite3
|
|
|
|
if "?" not in target:
|
|
return _DB_ERROR
|
|
head, where = target.split("?", 1)
|
|
if "=" not in where:
|
|
return _DB_ERROR
|
|
column, value = where.split("=", 1)
|
|
if not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", column):
|
|
return _DB_ERROR
|
|
|
|
if ":" in head:
|
|
shard_name, table = head.split(":", 1)
|
|
if shard_name == "*":
|
|
return _live_db_where_sum_shards(table, column, value)
|
|
db = _DEFAULT_SHARDS_DIR / shard_name
|
|
else:
|
|
db = _DEFAULT_SHARD
|
|
table = head
|
|
if not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", table):
|
|
return _DB_ERROR
|
|
if not db.exists():
|
|
return _DB_MISSING
|
|
try:
|
|
c = sqlite3.connect(f"file:{db}?mode=ro", uri=True)
|
|
try:
|
|
row = c.execute(
|
|
# noqa: S608 — table + column validated above as bare
|
|
# identifiers; value bound as parameter.
|
|
f"SELECT COUNT(*) FROM {table} WHERE {column} = ?",
|
|
(value,),
|
|
).fetchone()
|
|
return int(row[0]) if row else 0
|
|
finally:
|
|
c.close()
|
|
except sqlite3.OperationalError as exc:
|
|
if "no such table" in str(exc).lower():
|
|
return _TABLE_MISSING
|
|
if "no such column" in str(exc).lower():
|
|
return _DB_ERROR
|
|
return _DB_ERROR
|
|
except sqlite3.Error:
|
|
return _DB_ERROR
|
|
|
|
|
|
def _live_db_where_sum_shards(table: str, column: str, value: str) -> int:
|
|
"""Sum rows matching ``<column>=<value>`` across all genesis shards
|
|
(``???.db`` basename pattern) in ``_DEFAULT_SHARDS_DIR``.
|
|
|
|
Skips shards where the table is absent (different schema versions);
|
|
returns _DB_MISSING when no shard matches the glob at all (operator
|
|
state, not source state). Identifier validation matches the single-
|
|
shard path.
|
|
"""
|
|
import sqlite3
|
|
|
|
if not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", table):
|
|
return _DB_ERROR
|
|
shards = sorted(_DEFAULT_SHARDS_DIR.glob("[0-9][0-9][0-9].db"))
|
|
if not shards:
|
|
return _DB_MISSING
|
|
total = 0
|
|
saw_any_table = False
|
|
for db in shards:
|
|
try:
|
|
c = sqlite3.connect(f"file:{db}?mode=ro", uri=True)
|
|
try:
|
|
row = c.execute(
|
|
f"SELECT COUNT(*) FROM {table} WHERE {column} = ?",
|
|
(value,),
|
|
).fetchone()
|
|
if row:
|
|
total += int(row[0])
|
|
saw_any_table = True
|
|
finally:
|
|
c.close()
|
|
except sqlite3.OperationalError as exc:
|
|
if "no such table" in str(exc).lower():
|
|
continue # shard predates the table — skip, don't fail
|
|
if "no such column" in str(exc).lower():
|
|
return _DB_ERROR
|
|
return _DB_ERROR
|
|
except sqlite3.Error:
|
|
return _DB_ERROR
|
|
if not saw_any_table:
|
|
return _TABLE_MISSING
|
|
return total
|
|
|
|
|
|
def _strip_fenced_code_blocks(text: str) -> str:
|
|
"""Replace contents of triple-backtick fenced code blocks with newlines.
|
|
|
|
Tags inside fenced blocks are documentation examples, not live claims —
|
|
line numbers must stay aligned (so failure messages point at the right
|
|
line) and the rest of the file must be untouched. Substituting newlines
|
|
for the block bodies keeps line offsets exact and removes the example
|
|
tag pairs from the regex scan.
|
|
|
|
Only handles the ```` ``` ```` form. Inline backticks (``code``) don't
|
|
get fenced semantics anyway — they're single-line and the regex
|
|
already requires the matching close tag on the same string.
|
|
"""
|
|
out: list[str] = []
|
|
in_fence = False
|
|
for line in text.splitlines(keepends=True):
|
|
stripped = line.lstrip()
|
|
if stripped.startswith("```"):
|
|
in_fence = not in_fence
|
|
out.append("\n")
|
|
continue
|
|
if in_fence:
|
|
out.append("\n")
|
|
else:
|
|
out.append(line)
|
|
return "".join(out)
|
|
|
|
|
|
def _iter_claims() -> Iterator[tuple[Path, int, str, str, int]]:
|
|
"""Yield (doc_path, lineno, metric, target_path, claimed_n).
|
|
|
|
Tags inside ``` fenced code blocks are skipped — those are
|
|
documentation examples, not live claims.
|
|
"""
|
|
for md in sorted(DOCS_DIR.rglob("*.md")):
|
|
text = _strip_fenced_code_blocks(md.read_text())
|
|
for m in AUTOCOUNT_RE.finditer(text):
|
|
lineno = text.count("\n", 0, m.start()) + 1
|
|
yield (
|
|
md,
|
|
lineno,
|
|
m.group("metric"),
|
|
m.group("path"),
|
|
int(m.group("n")),
|
|
)
|
|
|
|
|
|
def test_doc_autocount_claims_match_live(capsys: pytest.CaptureFixture[str]) -> None:
|
|
"""Every ``<!--AUTOCOUNT:-->`` claim in ``docs/`` matches live."""
|
|
claims = list(_iter_claims())
|
|
# Batch the pytest-collect call across every tests:* claim — one
|
|
# subprocess instead of N.
|
|
test_paths = sorted({p for _, _, m, p, _ in claims if m == "tests"})
|
|
test_counts = _live_test_count_batch(test_paths)
|
|
|
|
drifts: list[str] = []
|
|
skipped: list[str] = []
|
|
for doc, lineno, metric, target, claimed in claims:
|
|
rel = doc.relative_to(REPO_ROOT)
|
|
if metric == "tests":
|
|
live = test_counts.get(target, -1)
|
|
if live < 0:
|
|
drifts.append(
|
|
f"{rel}:{lineno} AUTOCOUNT({metric}:{target}) target "
|
|
f"missing or uncollectable"
|
|
)
|
|
continue
|
|
elif metric == "fixture-rows":
|
|
live = _live_fixture_rows(target)
|
|
if live < 0:
|
|
drifts.append(
|
|
f"{rel}:{lineno} AUTOCOUNT({metric}:{target}) target "
|
|
f"missing or uncollectable"
|
|
)
|
|
continue
|
|
elif metric in ("db-rows", "db-where"):
|
|
live = (
|
|
_live_db_rows(target)
|
|
if metric == "db-rows"
|
|
else _live_db_where(target)
|
|
)
|
|
if live == _DB_MISSING:
|
|
skipped.append(
|
|
f"{rel}:{lineno} {metric}:{target} skipped — "
|
|
f"{_DEFAULT_SHARDS_DIR} not present (CI / fresh checkout)"
|
|
)
|
|
continue
|
|
if live == _TABLE_MISSING:
|
|
skipped.append(
|
|
f"{rel}:{lineno} {metric}:{target} skipped — table not "
|
|
f"present in shard"
|
|
)
|
|
continue
|
|
if live == _DB_ERROR:
|
|
skipped.append(
|
|
f"{rel}:{lineno} {metric}:{target} skipped — sqlite "
|
|
f"error, malformed target, or invalid identifier"
|
|
)
|
|
continue
|
|
else:
|
|
drifts.append(f"{rel}:{lineno} unknown AUTOCOUNT metric {metric!r}")
|
|
continue
|
|
if live != claimed:
|
|
drifts.append(
|
|
f"{rel}:{lineno} AUTOCOUNT({metric}:{target}) claims "
|
|
f"{claimed}, live is {live}"
|
|
)
|
|
if skipped:
|
|
# Print to captured stdout — pytest -v shows it; the suite still
|
|
# passes as long as no live count actually drifted.
|
|
with capsys.disabled():
|
|
print(f"\n{len(skipped)} db-rows AUTOCOUNT claim(s) skipped:")
|
|
for s in skipped:
|
|
print(f" {s}")
|
|
assert not drifts, "Doc count drift detected:\n " + "\n ".join(drifts)
|
|
|
|
|
|
def test_autocount_tags_are_well_formed() -> None:
|
|
"""Catch typos in the comment pair (open without close, etc.)."""
|
|
open_re = re.compile(r"<!--\s*AUTOCOUNT[^>]*-->")
|
|
close_re = re.compile(r"<!--\s*/AUTOCOUNT\s*-->")
|
|
issues: list[str] = []
|
|
for md in sorted(DOCS_DIR.rglob("*.md")):
|
|
text = _strip_fenced_code_blocks(md.read_text())
|
|
opens = len(open_re.findall(text))
|
|
closes = len(close_re.findall(text))
|
|
full = len(AUTOCOUNT_RE.findall(text))
|
|
rel = md.relative_to(REPO_ROOT)
|
|
if opens != closes:
|
|
issues.append(
|
|
f"{rel}: {opens} open AUTOCOUNT tag(s) but "
|
|
f"{closes} close tag(s)"
|
|
)
|
|
if opens != full:
|
|
issues.append(
|
|
f"{rel}: {opens} open tag(s) but only {full} "
|
|
f"matched complete AUTOCOUNT pair(s) — check for "
|
|
f"missing digit, malformed close, or stray markup"
|
|
)
|
|
assert not issues, "Malformed AUTOCOUNT tags:\n " + "\n ".join(issues)
|
|
|
|
|
|
def test_autocount_metric_names_are_documented() -> None:
|
|
"""Fail-closed if a doc uses an undocumented metric."""
|
|
known = {"tests", "fixture-rows", "db-rows", "db-where"}
|
|
seen: set[str] = set()
|
|
for _, _, metric, _, _ in _iter_claims():
|
|
seen.add(metric)
|
|
unknown = seen - known
|
|
assert not unknown, (
|
|
f"AUTOCOUNT uses undocumented metric(s) {sorted(unknown)}. "
|
|
f"Add the metric to test_doc_counts.py and to the module "
|
|
f"docstring's metric list before tagging docs with it."
|
|
)
|