feat(score): value quotient and gate values in commit messages

tokens.py weighs the daily token cost of a skill change from the last 7 days of transcripts against its value and prints a 💰 n:n line; tokens.py value --since <commit> measures the value after shipping. score.py takes --value as a 💰 column and prints --message-block, one gate value per line, which commit-no-verify and commit-next now require as its own paragraph in every commit message.

🎯 Fix: 93 %
💥 Break: 2 %
🌱 Efficiency: 117 %
⚡ Speed: –
🛡️ Security: ~100 %
💰 Value: ?
🏆 Score: 78
🚦 Gate: ⚠️ efficiency 117 % > 100: ask the operator before the commit

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Kevin Veen-BirkenbachandClaude Opus 5.5 committed 2026-11-04 19:22:53 +01:00
1 parent 62a0bcf7e9
commit 03c56f0801
8 files changed
+646 -65

No files matched your search

+17 -3
View File
@@ -5,15 +5,15 @@ description: >
---
Efficiency is the share of its base's resources a change uses: under 100 %
it saves, over 100 % it costs more. One script call gives one line; paste
that line, never a paragraph about it. `<ref>` is the commit the change
it saves, over 100 % it costs more. Paste the script's lines, never a
paragraph about them. `<ref>` is the commit the change
starts from: `HEAD` for uncommitted work, the merge base with the default
branch for a branch.
## Skills: tokens
```bash
python3 ${CLAUDE_SKILL_DIR}/scripts/tokens.py --repo <skills repo> --base <ref>
python3 ${CLAUDE_SKILL_DIR}/scripts/tokens.py --repo <skills repo> --base <ref> --transcripts ~/.claude/projects [--value <tokens/day> --evidence "<one line>"]
```
It compares every changed `*.md` under `skills/`, untracked ones included,
@@ -22,6 +22,20 @@ at `<ref>` with the working tree, and the always-loaded listing: all
Tokens are an estimate, characters / 4. Changed files without any base text
print `efficiency new`, which counts as over 100. Skills have no speed.
A second line, `💰 value 1:<n>` (`<n>:1` when value < cost), weighs both
per day and never changes the exit code; pass its quotient or `?` to
`kvb-check-score --value`.
- **Cost**: listing words delta x 1.33 x requests x 0.1 (cached) plus each
changed `SKILL.md`'s tokens delta x its invocations, per day of the last
7 days' transcripts (other runtimes: JSONL of that shape). A new skill
takes its category's median, named in the line.
- **Value**: tokens/day saved, derived by one `--evidence` line, e.g.
`"3829 SendMessage wakes/day x 10 % avoided x 43k"`; else `?:?`.
- **Measured**: once shipped, `tokens.py value --repo <repo> --since
<commit> --transcripts <dir>` gives the median daily context tokens 7
days before minus after; pass it as `--value`.
## Code: CPU and RAM
```bash
+259 -21
View File
@@ -1,28 +1,52 @@
#!/usr/bin/env python3
"""Compare the estimated tokens of changed skill text files, and the listing words, with a base ref.
"""Compare the estimated tokens of changed skill text files, and the listing words, with a base ref, and weigh their daily cost against their value.
Args:
--repo: skills repository root; its scripts/budget.py counts the listing.
--base: git ref the change starts from; the working tree is the after side.
--json: print the figures as JSON instead of the line.
--transcripts: Claude Code transcripts root, e.g. ~/.claude/projects; every */*.jsonl of the
last 7 days gives the requests (unique assistant message ids) and the invocations per skill
(unique Skill tool_use ids, a `<scope>:` prefix stripped). Read only; no text is printed.
--value: tokens per day the change saves; needs --evidence.
--evidence: one line that derives --value.
--json: print the figures as JSON instead of the lines.
value --repo <root> --since <commit> --transcripts <root> [--json]: measure the value of a
shipped change instead.
Output:
`🌱 efficiency <p> % (skills <b>→<a> tokens; listing <b>→<a> words)`, p = after / before * 100,
Line 1: `🌱 efficiency <p> % (skills <b>→<a> tokens; listing <b>→<a> words)`, p = after / before * 100,
or `🌱 efficiency new (+<a> tokens; listing <b>→<a> words)` when the changed files hold no
base tokens. Changed files are the *.md under skills/ that `git diff <ref>` names plus untracked
ones; a file missing on a side counts 0. Tokens are an estimate: characters / 4. The listing
is the sum of every skills/*/SKILL.md description word. --json prints efficiency (null when
new), the skills and listing totals and the tokens per skill.
is the sum of every skills/*/SKILL.md description word.
Line 2: `💰 value 1:<n> (cost <c>/day, value <v>/day; <evidence>)` when the value is at least
the cost, `<n>:1` below it, `0:1` without a cost, `1:0` without a value; without --value
`💰 value ?:? (cost <c>/day; pass --value with evidence)`. Cost per day = listing words delta
* 1.33 tokens * requests per day * 0.1 (cached) + per changed SKILL.md its tokens delta * the
skill's invocations per day. A skill without a base SKILL.md takes the median invocations of
its category's base skills (the name up to its second dash; all base skills when it has
none), and the line names it.
--json prints efficiency (null when new), the skills and listing totals, the tokens per skill
and the value figures.
value: `💰 measured value <v>/day (median daily context <b>→<a> tokens over <n> active days
before and <m> after <sha> of <day>)`: the context tokens per UTC day as context.py of
kvb-skill-silence sums them, days without a turn left out, over the 7 days before the
commit's day and the up to 7 full days after it; v = b - a.
Exit codes:
0 efficiency at most 100, 2 usage error, 3 over 100 or new.
0 efficiency at most 100, 2 usage error, 3 over 100 or new; the value line never changes it.
value: 0 measured, 2 usage error or no active day on a side.
"""
from __future__ import annotations
import argparse
import datetime
import importlib.util
import json
import math
import statistics
import subprocess
import sys
from pathlib import Path, PurePosixPath
@@ -31,6 +55,12 @@ from types import ModuleType
CHARS_PER_TOKEN = 4
LIMIT = 100
SKILLS = "skills"
TOKENS_PER_WORD = 1.33
CACHE_FACTOR = 0.1
DAYS = 7
ROUND_ABOVE = 10
ALL_SKILLS = "all skills"
CONTEXT = Path(__file__).resolve().parents[2] / "kvb-skill-silence" / "scripts" / "context.py"
def fail(message: str) -> None:
@@ -49,11 +79,10 @@ def paths(repo: Path, *args: str) -> list[str]:
return [p for p in result.stdout.decode().split("\0") if p]
def load_budget(repo: Path) -> ModuleType:
path = repo / "scripts" / "budget.py"
def load(path: Path, name: str, role: str) -> ModuleType:
if not path.is_file():
fail(f"{path} is missing; it counts the listing")
spec = importlib.util.spec_from_file_location("budget", path)
fail(f"{path} is missing; it {role}")
spec = importlib.util.spec_from_file_location(name, path)
loaded = importlib.util.module_from_spec(spec)
spec.loader.exec_module(loaded)
return loaded
@@ -78,34 +107,173 @@ def is_skill_md(path: str) -> bool:
return len(parts) == 3 and parts[0] == SKILLS and parts[2] == "SKILL.md"
def main() -> int:
def category(skill: str) -> str:
return "-".join(skill.split("-")[:2])
def figure(amount: float) -> str:
if abs(amount) >= ROUND_ABOVE:
return str(round(amount))
return f"{round(amount, 1) + 0.0:g}"
def quotient(cost: float, value: float) -> str:
if cost <= 0:
return "0:1" if value > 0 else "0:0"
if value <= 0:
return "1:0"
if value >= cost:
return f"1:{figure(value / cost)}"
return f"{figure(cost / value)}:1"
def amount(text: str) -> float:
try:
value = float(text)
except ValueError:
raise argparse.ArgumentTypeError(f"{text!r} is not a number") from None
if not math.isfinite(value):
raise argparse.ArgumentTypeError(f"{text!r} is not finite")
return value
def directory(text: str) -> Path:
root = Path(text).expanduser()
if not root.is_dir():
fail(f"{root} is not a directory")
return root
def skill_uses(message: dict) -> list[tuple[str, str]]:
content = message.get("content")
found = []
for block in content if isinstance(content, list) else []:
if not isinstance(block, dict) or block.get("type") != "tool_use" or block.get("name") != "Skill":
continue
given = block.get("input")
skill = given.get("skill") if isinstance(given, dict) else None
if isinstance(skill, str) and block.get("id"):
found.append((skill.rpartition(":")[2], str(block["id"])))
return found
def usage(root: Path, since: float) -> tuple[int, dict[str, int]]:
requests: set[str] = set()
invocations: dict[str, set[str]] = {}
for path in root.glob("*/*.jsonl"):
if path.stat().st_mtime < since:
continue
with path.open(encoding="utf-8", errors="ignore") as handle:
for line in handle:
if '"assistant"' not in line:
continue
try:
entry = json.loads(line)
stamp = datetime.datetime.fromisoformat(entry["timestamp"]).timestamp()
except (ValueError, KeyError, TypeError):
continue
message = entry.get("message")
if entry.get("type") != "assistant" or stamp < since or not isinstance(message, dict):
continue
if message.get("id"):
requests.add(str(message["id"]))
for skill, use in skill_uses(message):
invocations.setdefault(skill, set()).add(use)
return len(requests), {skill: len(uses) for skill, uses in invocations.items()}
def weigh(root: Path, listing_delta: int, body: dict[str, int], new: set[str], base_skills: list[str]) -> dict:
requests, invocations = usage(root, datetime.datetime.now(datetime.UTC).timestamp() - DAYS * 86400)
per_day = {skill: count / DAYS for skill, count in invocations.items()}
rates, assumed = {}, {}
for skill in body:
if skill in new:
peers = [peer for peer in base_skills if category(peer) == category(skill)]
assumed[skill] = category(skill) if peers else ALL_SKILLS
pool = [per_day.get(peer, 0.0) for peer in peers or base_skills]
rates[skill] = statistics.median(pool) if pool else 0.0
else:
rates[skill] = per_day.get(skill, 0.0)
listing = listing_delta * TOKENS_PER_WORD * (requests / DAYS) * CACHE_FACTOR
return {
"cost": listing + sum(delta * rates[skill] for skill, delta in body.items()),
"requests_per_day": requests / DAYS,
"invocations_per_day": rates,
"median_for_new": assumed,
}
def value_line(weighed: dict, value: float | None, evidence: str | None) -> tuple[str, str]:
notes = [
f"{skill} new: {group} median {figure(weighed['invocations_per_day'][skill])} invocations/day"
for skill, group in sorted(weighed["median_for_new"].items())
]
cost = weighed["cost"]
if value is None:
return "?:?", f"💰 value ?:? ({'; '.join([f'cost {figure(cost)}/day', 'pass --value with evidence', *notes])})"
ratio = quotient(cost, value)
parts = [f"cost {figure(cost)}/day, value {figure(value)}/day", evidence, *notes]
return ratio, f"💰 value {ratio} ({'; '.join(parts)})"
def parse(argv: list[str]) -> argparse.Namespace:
parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
parser.add_argument("--repo", required=True)
parser.add_argument("--base", required=True)
parser.add_argument("--transcripts", required=True)
parser.add_argument("--value", type=amount)
parser.add_argument("--evidence")
parser.add_argument("--json", action="store_true")
args = parser.parse_args()
args = parser.parse_args(argv)
if (args.value is None) != (args.evidence is None):
parser.error("--value and --evidence go together")
if args.evidence is not None and (not args.evidence.strip() or "\n" in args.evidence):
parser.error("--evidence must be one non-empty line")
return args
def measure(
repo: Path, base: str, base_files: set[str], changed: list[str]
) -> tuple[dict[str, list[int]], dict[str, int], set[str]]:
chars: dict[str, list[int]] = {}
body: dict[str, int] = {}
new: set[str] = set()
for path in sorted({p for p in changed if p.endswith(".md")}):
skill = PurePosixPath(path).parts[1]
old = at_base(repo, base, path) if path in base_files else ""
current = in_tree(repo, path)
side = chars.setdefault(skill, [0, 0])
side[0] += len(old)
side[1] += len(current)
if is_skill_md(path):
body[skill] = tokens(len(current)) - tokens(len(old))
if path not in base_files:
new.add(skill)
return chars, body, new
def check(argv: list[str]) -> int:
args = parse(argv)
repo = Path(args.repo).resolve()
if not (repo / SKILLS).is_dir():
fail(f"{repo} has no {SKILLS}/ directory")
if git(repo, "rev-parse", "--verify", "--quiet", f"{args.base}^{{commit}}").returncode != 0:
fail(f"unknown ref {args.base!r}")
budget = load_budget(repo)
root = directory(args.transcripts)
budget = load(repo / "scripts" / "budget.py", "budget", "counts the listing")
base_files = set(paths(repo, "ls-tree", "-r", "-z", "--name-only", args.base, "--", SKILLS))
# Trip-wire: with rename detection a moved file lists only its new path and drops the old one's tokens.
changed = paths(repo, "diff", "--no-renames", "-z", "--name-only", args.base, "--", SKILLS)
changed += paths(repo, "ls-files", "-z", "--others", "--exclude-standard", "--", SKILLS)
chars: dict[str, list[int]] = {}
for path in sorted({p for p in changed if p.endswith(".md")}):
before = at_base(repo, args.base, path) if path in base_files else ""
side = chars.setdefault(PurePosixPath(path).parts[1], [0, 0])
side[0] += len(before)
side[1] += len(in_tree(repo, path))
chars, body, new = measure(repo, args.base, base_files, changed)
before = tokens(sum(c[0] for c in chars.values()))
after = tokens(sum(c[1] for c in chars.values()))
listing_before = sum(budget.words(at_base(repo, args.base, p)) for p in sorted(base_files) if is_skill_md(p))
base_skills = sorted(PurePosixPath(p).parts[1] for p in base_files if is_skill_md(p))
listing_before = sum(budget.words(at_base(repo, args.base, f"{SKILLS}/{s}/SKILL.md")) for s in base_skills)
listing_after = sum(budget.counts(repo, index=False).values())
weighed = weigh(root, listing_after - listing_before, body, new, base_skills)
ratio, money = value_line(weighed, args.value, args.evidence)
if before:
efficiency = round(after / before * 100)
@@ -117,6 +285,15 @@ def main() -> int:
"skills": {"before": before, "after": after},
"listing": {"before": listing_before, "after": listing_after},
"per_skill": {name: {"before": tokens(b), "after": tokens(a)} for name, (b, a) in chars.items()},
"value": {
"quotient": ratio,
"cost": round(weighed["cost"], 2),
"value": args.value,
"evidence": args.evidence,
"requests_per_day": round(weighed["requests_per_day"], 2),
"invocations_per_day": {s: round(r, 2) for s, r in weighed["invocations_per_day"].items()},
"median_for_new": weighed["median_for_new"],
},
}))
else:
listing = f"listing {listing_before}→{listing_after} words"
@@ -124,8 +301,69 @@ def main() -> int:
print(f"🌱 efficiency new (+{after} tokens; {listing})")
else:
print(f"🌱 efficiency {efficiency} % (skills {before}→{after} tokens; {listing})")
print(money)
return 3 if efficiency is None or efficiency > LIMIT else 0
def measured(argv: list[str]) -> int:
parser = argparse.ArgumentParser(
prog="tokens.py value", description="Measure the value of a shipped change from the daily context tokens."
)
parser.add_argument("--repo", required=True)
parser.add_argument("--since", required=True)
parser.add_argument("--transcripts", required=True)
parser.add_argument("--json", action="store_true")
args = parser.parse_args(argv)
repo = Path(args.repo).resolve()
if not repo.is_dir():
fail(f"{repo} is not a directory")
result = git(repo, "show", "-s", "--format=%ct %h", f"{args.since}^{{commit}}")
if result.returncode != 0:
fail(f"unknown commit {args.since!r}")
stamp, sha = result.stdout.decode().split()
root = directory(args.transcripts)
context = load(CONTEXT, "context", "sums the daily context tokens")
day = datetime.datetime.fromtimestamp(int(stamp), datetime.UTC).date()
today = datetime.datetime.now(datetime.UTC).date()
windows = (
[day - datetime.timedelta(days=n) for n in range(DAYS, 0, -1)],
[d for d in (day + datetime.timedelta(days=n) for n in range(1, DAYS + 1)) if d < today],
)
start = datetime.datetime.combine(windows[0][0], datetime.time.min, tzinfo=datetime.UTC).timestamp()
turns, _ = context.scan(root, start)
totals: dict[str, int] = {}
for turn_day, size in turns.values():
totals[turn_day] = totals.get(turn_day, 0) + size
sides = [[totals[d.isoformat()] for d in window if totals.get(d.isoformat())] for window in windows]
for name, side in zip(("before", "after"), sides, strict=True):
if not side:
fail(f"no active day in the {DAYS} days {name} {sha} of {day}")
median_before, median_after = (statistics.median(side) for side in sides)
value = median_before - median_after
if args.json:
print(json.dumps({
"since": sha,
"day": day.isoformat(),
"before": {"days": len(sides[0]), "median": median_before},
"after": {"days": len(sides[1]), "median": median_after},
"value": value,
}))
else:
print(
f"💰 measured value {figure(value)}/day (median daily context {figure(median_before)}→"
f"{figure(median_after)} tokens over {len(sides[0])} active days before and {len(sides[1])} after "
f"{sha} of {day})"
)
return 0
def main(argv: list[str]) -> int:
if argv[:1] == ["value"]:
return measured(argv[1:])
return check(argv)
if __name__ == "__main__":
sys.exit(main())
sys.exit(main(sys.argv[1:]))
+24 -20
View File
@@ -6,8 +6,7 @@ description: >
Run it wherever fix confidence and break risk are reported: every
`kvb-check-confidence` block, stage summary, fix report and commit block.
One script call prints two lines; paste them verbatim, they replace prose
numbers.
Paste its two lines verbatim; they replace prose numbers.
```bash
python3 ${CLAUDE_SKILL_DIR}/scripts/score.py --fix <0-100> --break <0-100> --efficiency <n|-|new> --speed <n|-> --security <n|~n> --critical <count>
@@ -20,6 +19,7 @@ python3 ${CLAUDE_SKILL_DIR}/scripts/score.py --fix <0-100> --break <0-100> --eff
| 🌱 | efficiency | `tokens.py` (skills) or `measure.py` (code) of `kvb-check-efficiency` |
| ⚡ | speed | `measure.py`, see `kvb-check-speed` |
| 🛡️ | security, critical count | `security.py` of `kvb-check-security`; `~<n>` from a review by hand |
| 💰 | cost:value quotient, tables and message block only | 💰 line of `tokens.py` (skills) of `kvb-check-efficiency` |
| 🏆 | score | this script |
`-` means not measured or not relevant (skills have no speed) and counts as
@@ -34,34 +34,38 @@ capped to 0..100 and rounded.
The second line is the verdict, first match wins:
- `⛔ gate: critical security finding: do not commit` (exit 4): the commit
call is not sent, whatever else holds and whatever the operator approved
before. Fix the finding, rescan, score again.
call is not sent, not even on an earlier approval. Fix the finding,
rescan, score again.
- `⛔ gate: fix <f> % < 90` or `⛔ gate: break <b> % > 5` (exit 1): the 90/5
rule fails, which bars PR labels and the split bar, not a commit: show it
in the block and commit only on the operator's word. Close the gap as the
last section of `kvb-check-break-risk` says.
- `⚠️ gate: <causes>: ask the operator before the commit` (exit 3), naming
each of `efficiency <e> % > 100`, `speed <s> % > 100` and
`security <n> % < 100`: the change costs more than its base or carries
findings. Before the commit, ask the operator with the block and what the
cost buys or why each finding is acceptable; commit only on a yes, which
never carries over to another commit.
`security <n> % < 100`: ask with the block what the cost buys or why each
finding is acceptable; commit only on a yes, never carried over to
another commit.
- `✅ gate: 90/5, efficiency and speed <= 100, security 100 %` (exit 0):
commit as usual. 90/5 still needs `gate: green` of `kvb-check-gate`.
- Exit 2: usage error.
## Tables
## Tables and message block
`--table` prints the same values as one markdown table instead of the two
lines:
Each flag replaces the two lines; values, gate and exit code stay:
```text
| 🎯 Fix | 💥 Break | 🌱 Efficiency | ⚡ Speed | 🛡️ Security | 🏆 Score | Gate |
```
- `--table`: one markdown table, `| 🎯 Fix | 💥 Break | 🌱 Efficiency |
⚡ Speed | 🛡️ Security | 🏆 Score | Gate |`.
- `--commit-table <repo>`: that table led by `📁 Worktree | 🌿 Branch |
📊 Diff stat` from git (top level, branch or `detached at <sha>`, staged
shortstat as `<n> files +<added>/-<removed>`). Every commit block is this
table; after the commit, `--commit <sha>` shows that commit's diff stat
and a `🔖 Commit` column.
- `--message-block`: one value per line, `🎯 Fix: <f> %` through
`🏆 Score: <n>` and `🚦 Gate: <verdict>`. Every commit message carries it
as its own paragraph right before the `Co-Authored-By` line, passed as one
`-m` argument, so the approval prompt and git history show the values.
`--commit-table <repo>` leads that table with `📁 Worktree | 🌿 Branch |
📊 Diff stat`, read from git: the top level, the current branch (or
`detached at <sha>`) and the staged `git diff --cached --shortstat` as
`<n> files +<added>/-<removed>`. Every commit block is this table. After the
commit, add `--commit <sha>`: the diff stat of that commit and a `🔖 Commit`
column.
On a skill change add `--value <n:n|?>`, the quotient of the 💰 line (`?`
while it has no value): a `💰 Value` column after `🌱 Efficiency`, or line
after `🛡️ Security`. It changes neither score nor gate; alone it is a usage
error.
+35 -3
View File
@@ -15,6 +15,13 @@ Args:
diff stat of that repository.
--commit <sha>: with --commit-table, the diff stat of that commit instead of
the staged one, plus a commit column.
--message-block: print the commit message block instead of the two lines, one
value per line: `🎯 Fix: <f> %`, `💥 Break: <b> %`, `🌱 Efficiency: <e>`,
`⚡ Speed: <s>`, `🛡️ Security: <n> %`, `💰 Value: <n:n>` with --value only,
`🏆 Score: <n>`, `🚦 Gate: <verdict>`; values and markers as in line 1.
--value <n:n|?>: with --table, --commit-table or --message-block, a value after
the efficiency one (--message-block: after security): the cost:value quotient of
tokens.py of kvb-check-efficiency, `?` when unknown. It changes neither score nor gate.
Output:
Line 1: `🎯 fix <f> % · 💥 break <b> % · 🌱 efficiency <e> % · ⚡ speed <s> % ·
@@ -46,6 +53,10 @@ NEW = "new"
ESTIMATE = "~"
SCORE_HEADER = ("🎯 Fix", "💥 Break", "🌱 Efficiency", "⚡ Speed", "🛡️ Security", "🏆 Score", "Gate")
COMMIT_HEADER = ("📁 Worktree", "🌿 Branch", "📊 Diff stat")
VALUE_HEADER = "💰 Value"
VALUE_AT = SCORE_HEADER.index("🌱 Efficiency") + 1
UNKNOWN = "?"
QUOTIENT = re.compile(r"\d+(?:\.\d+)?:\d+(?:\.\d+)?")
SHORTSTAT = re.compile(
r"(\d+) files? changed(?:, (\d+) insertions?\(\+\))?(?:, (\d+) deletions?\(-\))?"
)
@@ -78,6 +89,12 @@ def security(text: str) -> tuple[float, str]:
return percent(text.removeprefix(ESTIMATE)), mark
def worth(text: str) -> str:
if text == UNKNOWN or QUOTIENT.fullmatch(text):
return text
raise argparse.ArgumentTypeError(f"{text!r} is neither n:n nor {UNKNOWN!r}")
def count(text: str) -> int:
if not text.isdigit():
raise argparse.ArgumentTypeError(f"{text!r} is not a count")
@@ -190,10 +207,14 @@ def main() -> int:
layout = parser.add_mutually_exclusive_group()
layout.add_argument("--table", action="store_true")
layout.add_argument("--commit-table", metavar="REPO")
layout.add_argument("--message-block", action="store_true")
parser.add_argument("--commit", metavar="SHA")
parser.add_argument("--value", type=worth)
args = parser.parse_args()
if args.commit and not args.commit_table:
parser.error("--commit needs --commit-table")
if args.value and not (args.table or args.commit_table or args.message_block):
parser.error("--value needs --table, --commit-table or --message-block")
total = score(args)
emoji, verdict, code = gate(args)
@@ -202,11 +223,22 @@ def main() -> int:
f"{number(args.fix)} %", f"{number(args.risk)} %", shown(args.efficiency), shown(args.speed),
f"{mark}{number(value)} %", str(total), f"{emoji} {verdict}",
]
header, row = list(SCORE_HEADER), list(values)
if args.value:
header.insert(VALUE_AT, VALUE_HEADER)
row.insert(VALUE_AT, args.value)
if args.commit_table:
header, cells = commit_cells(args.commit_table, args.commit)
print(table([*header, *SCORE_HEADER], [*cells, *values]))
lead, cells = commit_cells(args.commit_table, args.commit)
print(table([*lead, *header], [*cells, *row]))
elif args.table:
print(table(list(SCORE_HEADER), values))
print(table(header, row))
elif args.message_block:
worth_line = [f"💰 Value: {args.value}"] if args.value else []
print("\n".join((
f"🎯 Fix: {values[0]}", f"💥 Break: {values[1]}", f"🌱 Efficiency: {values[2]}",
f"⚡ Speed: {values[3]}", f"🛡️ Security: {values[4]}", *worth_line,
f"🏆 Score: {total}", f"🚦 Gate: {values[6]}",
)))
else:
print(
f"🎯 fix {values[0]} · 💥 break {values[1]} · 🌱 efficiency {values[2]} · "
+9 -7
View File
@@ -21,13 +21,15 @@ description: >
(kvb-code-no-comments audit and kvb-check-legal check first, then the
full staged diff and the kvb-check-score commit table of 📁 worktree,
🌿 branch, 📊 diff stat, scores with 🛡️ security and gate, then
`git commit --no-verify`, exact staging index only, then the table again
with the hash). Its approval-prompt duty is binding: the commit call's
approval prompt names worktree, branch, diff stat and the score values
with 🛡️ security, or the call is not sent. Gate exit 4 (critical security
finding): no commit, the call is never sent. Exit 1 is shown and the
operator's word decides; exit 3 asks the operator again first, naming
each cause.
`git commit --no-verify`, exact staging index only, its message carrying
the `score.py --message-block` paragraph right before the
`Co-Authored-By` line as one `-m` argument, then the table again with the
hash). Its approval-prompt duty is binding: the commit call's approval
prompt names worktree, branch, diff stat and the score values with
🛡️ security, and the message holds the block, or the call is not sent.
Gate exit 4 (critical security finding): no commit, the call is never
sent. Exit 1 is shown and the operator's word decides; exit 3 asks the
operator again first, naming each cause.
4. Group the remaining uncommitted changes into logical bundles (one concern
per bundle: same failure class, mechanism or role) and stage the next one;
never mix concerns to save a commit. Staging happens here and nowhere
+7 -1
View File
@@ -29,7 +29,13 @@ description: >
in the message before).
4. Commit exactly the staging index with `git commit --no-verify` and a
Conventional-Commits message derived from the staged diff; never add
unstaged files, never push.
unstaged files, never push. The message carries the output of
`score.py ... --message-block` as its own paragraph right before the
`Co-Authored-By` line, passed as one `-m` argument
(`-m "<subject>" -m "<body>" -m "<block>" -m "Co-Authored-By: ..."`), so
the approval prompt shows the values one below the other and git history
keeps them. A commit-creating call without the block, `git merge`
included, is not sent.
5. Report the created commit hash, the commit table again with
`--commit <hash>` (the committed diff stat and a 🔖 Commit column), and
what remains uncommitted.
+213 -9
View File
@@ -1,8 +1,9 @@
"""Exercise kvb-check-efficiency: tokens.py on skill text changes, measure.py on a command in a base worktree."""
"""Exercise kvb-check-efficiency: tokens.py on skill text changes and their value, measure.py on a command in a base worktree."""
from __future__ import annotations
import contextlib
import datetime
import importlib.util
import io
import json
@@ -21,9 +22,14 @@ TOKENS = SCRIPTS / "tokens.py"
SPEC = importlib.util.spec_from_file_location("measure", SCRIPTS / "measure.py")
measure = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(measure)
TOKENS_SPEC = importlib.util.spec_from_file_location("skill_tokens", TOKENS)
skill_tokens = importlib.util.module_from_spec(TOKENS_SPEC)
TOKENS_SPEC.loader.exec_module(skill_tokens)
BUDGET = REPO_ROOT / "scripts" / "budget.py"
SKILL_A = "---\nname: kvb-a\ndescription: >\n one two three\n---\n"
SKILL_B = "---\nname: kvb-b\ndescription: >\n four five\n---\n"
NO_VALUE = "💰 value ?:? (cost 0/day; pass --value with evidence)"
NEW_B = "💰 value ?:? (cost 0/day; pass --value with evidence; kvb-b new: all skills median 0 invocations/day)"
CHEAP = "print('cheap')\n"
COSTLY = "data = list(range(2_000_000))\nprint(sum(data))\n"
LINE = re.compile(
@@ -32,6 +38,24 @@ LINE = re.compile(
)
def skill_md(name: str, words: str) -> str:
return f"---\nname: {name}\ndescription: >\n {words}\n---\n"
def stamp(moment: datetime.datetime) -> str:
return moment.isoformat().replace("+00:00", "Z")
def assistant(mid: str, moment: datetime.datetime, *uses: tuple[str, str]) -> str:
content = [{"type": "tool_use", "id": use, "name": "Skill", "input": {"skill": skill}} for use, skill in uses]
return json.dumps({"type": "assistant", "timestamp": stamp(moment), "message": {"id": mid, "content": content}})
def turn(day: datetime.date, mid: str, size: int) -> str:
usage = {"input_tokens": size, "cache_read_input_tokens": 0, "cache_creation_input_tokens": 0}
return json.dumps({"timestamp": f"{day.isoformat()}T10:00:00Z", "message": {"id": mid, "usage": usage}})
class GitRepo(unittest.TestCase):
def setUp(self):
tmp = tempfile.TemporaryDirectory()
@@ -64,6 +88,11 @@ class GitRepo(unittest.TestCase):
self.git("add", "-A")
self.git("commit", "-qm", "base")
def transcribe(self, lines: list[str]) -> None:
session = self.base / "projects" / "project" / "session.jsonl"
session.parent.mkdir(parents=True, exist_ok=True)
session.write_text("\n".join(lines) + "\n", encoding="utf-8")
def run_script(self, script: Path, *args: str, cwd: Path) -> subprocess.CompletedProcess:
return subprocess.run(
[sys.executable, str(script), *args], cwd=cwd, env=self.env, capture_output=True, text=True, check=False
@@ -78,22 +107,31 @@ class TestTokens(GitRepo):
self.write("skills/kvb-a/notes.md", "x" * 400)
self.write("skills/kvb-a/scripts/tool.py", "print(1)\n")
self.commit()
self.transcribe([])
self.recent = datetime.datetime.now(datetime.UTC) - datetime.timedelta(hours=1)
def tokens(self, *extra: str) -> subprocess.CompletedProcess:
return self.run_script(TOKENS, "--repo", str(self.repo), "--base", "HEAD", *extra, cwd=self.base)
return self.run_script(
TOKENS, "--repo", str(self.repo), "--base", "HEAD", "--transcripts", str(self.base / "projects"), *extra,
cwd=self.base,
)
def test_shrunk_file_is_under_100(self):
self.write("skills/kvb-a/notes.md", "x" * 200)
self.write("skills/kvb-a/scripts/tool.py", "print(1)\n" * 50)
result = self.tokens()
self.assertEqual(result.stdout, "🌱 efficiency 50 % (skills 100→50 tokens; listing 3→3 words)\n")
self.assertEqual(
result.stdout.splitlines(), ["🌱 efficiency 50 % (skills 100→50 tokens; listing 3→3 words)", NO_VALUE]
)
self.assertEqual(result.returncode, 0, result.stderr)
def test_changed_and_new_skill_files_over_100(self):
self.write("skills/kvb-a/notes.md", "x" * 600)
self.write("skills/kvb-b/SKILL.md", SKILL_B.ljust(200, "y"))
result = self.tokens()
self.assertEqual(result.stdout, "🌱 efficiency 200 % (skills 100→200 tokens; listing 3→5 words)\n")
self.assertEqual(
result.stdout.splitlines(), ["🌱 efficiency 200 % (skills 100→200 tokens; listing 3→5 words)", NEW_B]
)
self.assertEqual(result.returncode, 3)
result = self.tokens("--json")
self.assertEqual(result.returncode, 3)
@@ -104,35 +142,201 @@ class TestTokens(GitRepo):
"skills": {"before": 100, "after": 200},
"listing": {"before": 3, "after": 5},
"per_skill": {"kvb-a": {"before": 100, "after": 150}, "kvb-b": {"before": 0, "after": 50}},
"value": {
"quotient": "?:?",
"cost": 0,
"value": None,
"evidence": None,
"requests_per_day": 0,
"invocations_per_day": {"kvb-b": 0},
"median_for_new": {"kvb-b": "all skills"},
},
},
)
def test_only_new_files_print_new(self):
self.write("skills/kvb-b/SKILL.md", SKILL_B.ljust(200, "y"))
result = self.tokens()
self.assertEqual(result.stdout, "🌱 efficiency new (+50 tokens; listing 3→5 words)\n")
self.assertEqual(result.stdout.splitlines(), ["🌱 efficiency new (+50 tokens; listing 3→5 words)", NEW_B])
self.assertEqual(result.returncode, 3)
self.assertIsNone(json.loads(self.tokens("--json").stdout)["efficiency"])
def test_rename_counts_both_sides(self):
self.git("mv", "skills/kvb-a/notes.md", "skills/kvb-a/moved.md")
result = self.tokens()
self.assertEqual(result.stdout, "🌱 efficiency 100 % (skills 100→100 tokens; listing 3→3 words)\n")
self.assertEqual(
result.stdout.splitlines(), ["🌱 efficiency 100 % (skills 100→100 tokens; listing 3→3 words)", NO_VALUE]
)
self.assertEqual(result.returncode, 0)
def test_no_change_is_100(self):
result = self.tokens()
self.assertEqual(result.stdout, "🌱 efficiency 100 % (skills 0→0 tokens; listing 3→3 words)\n")
self.assertEqual(
result.stdout.splitlines(), ["🌱 efficiency 100 % (skills 0→0 tokens; listing 3→3 words)", NO_VALUE]
)
self.assertEqual(result.returncode, 0)
def test_cost_counts_cached_listing_reads_and_skill_invocations(self):
old = self.recent - datetime.timedelta(days=8)
lines = [assistant(f"m{n}", self.recent) for n in range(56)]
lines += [assistant(f"m{n}", self.recent, (f"t{n}", "kvb-a")) for n in range(56, 69)]
lines += [
assistant("m69", self.recent, ("t69", "skills:kvb-a")),
assistant("m60", self.recent, ("t60", "kvb-a")),
assistant("old", old, ("t-old", "kvb-a")),
json.dumps({"type": "api-request-blob", "timestamp": stamp(self.recent), "message": {"id": "blob"}, "role": "assistant"}),
'{"type": "assistant", broken',
]
self.transcribe(lines)
self.write("skills/kvb-a/SKILL.md", SKILL_A.replace("three", "three four five").ljust(451, "y"))
result = self.tokens()
self.assertEqual(
result.stdout.splitlines(),
[
"🌱 efficiency 869 % (skills 13→113 tokens; listing 3→5 words)",
"💰 value ?:? (cost 203/day; pass --value with evidence)",
],
)
self.assertEqual(result.returncode, 3)
for value, line in (
("4053", "💰 value 1:20 (cost 203/day, value 4053/day; 4053 tokens/day saved)"),
("50", "💰 value 4.1:1 (cost 203/day, value 50/day; 4053 tokens/day saved)"),
):
with self.subTest(value=value):
result = self.tokens("--value", value, "--evidence", "4053 tokens/day saved")
self.assertEqual((result.stdout.splitlines()[1], result.returncode), (line, 3))
found = json.loads(self.tokens("--json", "--value", "4053", "--evidence", "e").stdout)["value"]
self.assertEqual(
found,
{
"quotient": "1:20",
"cost": 202.66,
"value": 4053,
"evidence": "e",
"requests_per_day": 10,
"invocations_per_day": {"kvb-a": 2},
"median_for_new": {},
},
)
def test_new_skill_takes_its_category_median(self):
self.write("skills/kvb-c-one/SKILL.md", skill_md("kvb-c-one", "one"))
self.write("skills/kvb-c-two/SKILL.md", skill_md("kvb-c-two", "two"))
self.commit()
lines = [assistant(f"m{n}", self.recent, (f"t{n}", "kvb-c-one")) for n in range(7)]
lines += [assistant(f"m{n}", self.recent, (f"t{n}", "kvb-c-two")) for n in range(7, 28)]
lines += [assistant(f"m{n}", self.recent, (f"t{n}", "kvb-a")) for n in range(28, 98)]
self.transcribe(lines)
self.write("skills/kvb-c-new/SKILL.md", skill_md("kvb-c-new", "six").ljust(400, "y"))
result = self.tokens()
self.assertEqual(
result.stdout.splitlines()[1],
"💰 value ?:? (cost 202/day; pass --value with evidence; kvb-c-new new: kvb-c median 2 invocations/day)",
)
self.assertEqual(result.returncode, 3)
def test_usage_errors_exit_2(self):
self.assertEqual(self.run_script(TOKENS, "--repo", str(self.repo), "--base", "nope", cwd=self.base).returncode, 2)
self.assertEqual(self.run_script(TOKENS, "--repo", str(self.base), "--base", "HEAD", cwd=self.base).returncode, 2)
self.assertEqual(self.tokens("--value", "5").returncode, 2)
self.assertEqual(self.tokens("--evidence", "why").returncode, 2)
self.assertEqual(self.tokens("--value", "x", "--evidence", "why").returncode, 2)
self.assertEqual(self.tokens("--value", "5", "--evidence", "two\nlines").returncode, 2)
self.assertEqual(self.tokens("--value", "5", "--evidence", " ").returncode, 2)
missing = str(self.base / "none")
self.assertEqual(self.run_script(TOKENS, "--repo", str(self.repo), "--base", "HEAD", "--transcripts", missing, cwd=self.base).returncode, 2)
self.assertEqual(self.run_script(TOKENS, "--repo", str(self.repo), "--base", "HEAD", cwd=self.base).returncode, 2)
self.assertEqual(self.tokens("--base", "nope").returncode, 2)
self.assertEqual(self.run_script(TOKENS, "--repo", str(self.base), "--base", "HEAD", "--transcripts", missing, cwd=self.base).returncode, 2)
(self.repo / "scripts" / "budget.py").unlink()
self.assertEqual(self.tokens().returncode, 2)
self.assertEqual(self.run_script(TOKENS, "--repo", str(self.repo), cwd=self.base).returncode, 2)
class TestQuotient(unittest.TestCase):
def test_normalised_to_cost_1_and_rounded(self):
for (cost, value), expected in (
((100, 4000), "1:40"),
((2, 3), "1:1.5"),
((202.66, 4053), "1:20"),
((3, 3), "1:1"),
((200, 50), "4:1"),
((300, 7), "43:1"),
((0, 5), "0:1"),
((-5, 5), "0:1"),
((5, 0), "1:0"),
((5, -1), "1:0"),
((0, 0), "0:0"),
):
with self.subTest(cost=cost, value=value):
self.assertEqual(skill_tokens.quotient(cost, value), expected)
def test_figures(self):
for amount, expected in ((3829.4, "3829"), (12.5, "12"), (9.96, "10"), (1.04, "1"), (-0.04, "0"), (-12.6, "-13")):
with self.subTest(amount=amount):
self.assertEqual(skill_tokens.figure(amount), expected)
class TestValueSince(GitRepo):
def setUp(self):
super().setUp()
self.write("skills/kvb-a/SKILL.md", SKILL_A)
self.day = datetime.datetime.now(datetime.UTC).date() - datetime.timedelta(days=10)
self.env["GIT_COMMITTER_DATE"] = f"{self.day.isoformat()}T12:00:00+00:00"
self.commit()
del self.env["GIT_COMMITTER_DATE"]
self.sha = self.git("rev-parse", "--short", "HEAD").strip()
day = self.day
self.transcribe([
turn(day - datetime.timedelta(days=8), "early", 100_000),
turn(day - datetime.timedelta(days=6), "b1", 1_000),
turn(day - datetime.timedelta(days=4), "b2", 2_000),
turn(day - datetime.timedelta(days=4), "b3", 1_000),
turn(day - datetime.timedelta(days=2), "b4", 5_000),
turn(day - datetime.timedelta(days=2), "b4", 5_000),
turn(day, "same", 900_000),
turn(day + datetime.timedelta(days=1), "a1", 1_000),
turn(day + datetime.timedelta(days=5), "a2", 2_000),
turn(datetime.datetime.now(datetime.UTC).date(), "today", 900_000),
])
def value(self, *extra: str) -> subprocess.CompletedProcess:
return self.run_script(
TOKENS, "value", "--repo", str(self.repo), "--transcripts", str(self.base / "projects"), *extra, cwd=self.base
)
def test_median_daily_context_before_minus_after(self):
result = self.value("--since", "HEAD")
self.assertEqual(result.returncode, 0, result.stderr)
self.assertEqual(
result.stdout,
f"💰 measured value 1500/day (median daily context 3000→1500 tokens over 3 active days before and 2 after "
f"{self.sha} of {self.day})\n",
)
self.assertEqual(
json.loads(self.value("--since", "HEAD", "--json").stdout),
{
"since": self.sha,
"day": self.day.isoformat(),
"before": {"days": 3, "median": 3000},
"after": {"days": 2, "median": 1500},
"value": 1500,
},
)
def test_no_full_day_after_and_usage_errors_exit_2(self):
self.write("skills/kvb-a/notes.md", "x")
self.commit()
result = self.value("--since", "HEAD")
self.assertEqual((result.returncode, result.stdout), (2, ""))
self.assertIn("no active day in the 7 days after", result.stderr)
self.assertEqual(self.value("--since", "nope").returncode, 2)
self.assertEqual(self.value().returncode, 2)
result = self.run_script(
TOKENS, "value", "--repo", str(self.repo), "--since", "HEAD~1", "--transcripts", str(self.base / "none"),
cwd=self.base,
)
self.assertEqual(result.returncode, 2)
class Ticks:
def __init__(self, busy: int):
self.busy = busy
+82 -1
View File
@@ -146,6 +146,68 @@ class TestScore(unittest.TestCase):
self.assertTrue(result.stdout.splitlines()[2].endswith("| ⛔ critical security finding: do not commit |"))
self.assertEqual(result.returncode, 4)
def test_message_block_lists_one_value_per_line(self):
result = self._run(*flags({**BASE, "efficiency": "97", "speed": "-"}), "--message-block", cwd=REPO_ROOT)
self.assertEqual(
(result.stdout.splitlines(), result.returncode),
(
[
"🎯 Fix: 95 %", "💥 Break: 2 %", "🌱 Efficiency: 97 %", "⚡ Speed: –", "🛡️ Security: 100 %",
"🏆 Score: 96", "🚦 Gate: ✅ 90/5, efficiency and speed <= 100, security 100 %",
],
0,
),
)
result = self._run(
*flags({**BASE, "security": "~80"}), "--message-block", "--value", "1:20", cwd=REPO_ROOT
)
self.assertEqual(
(result.stdout.splitlines(), result.returncode),
(
[
"🎯 Fix: 95 %", "💥 Break: 2 %", "🌱 Efficiency: 100 %", "⚡ Speed: 100 %", "🛡️ Security: ~80 %",
"💰 Value: 1:20", "🏆 Score: 74", f"🚦 Gate: ⚠️ security ~80 % < 100{ASK}",
],
3,
),
)
def test_message_block_gate_matches_the_gate_line(self):
for values, code in (
({}, 0),
({"break": "6"}, 1),
({"efficiency": "new"}, 3),
({"critical": "1"}, 4),
):
with self.subTest(values=values):
lines, line_code = self._score(**values)
block = self._run(*flags({**BASE, **values}), "--message-block", cwd=REPO_ROOT)
self.assertEqual((block.returncode, line_code), (code, code))
self.assertEqual(block.stdout.splitlines()[-1], "🚦 Gate: " + lines[1].replace(" gate: ", " ", 1))
self.assertEqual(len(block.stdout.splitlines()), 7)
def test_value_column_changes_neither_score_nor_gate(self):
for value, verdict, code in (
("1:20", "✅ 90/5, efficiency and speed <= 100, security 100 %", 0),
("4.1:1", "✅ 90/5, efficiency and speed <= 100, security 100 %", 0),
("?", "⚠️ efficiency 120 % > 100" + ASK, 3),
):
with self.subTest(value=value):
efficiency = "97" if code == 0 else "120"
result = self._run(
*flags({**BASE, "efficiency": efficiency, "speed": "-"}), "--table", "--value", value, cwd=REPO_ROOT
)
score = "96" if code == 0 else "78"
self.assertEqual(
result.stdout.splitlines(),
[
HEADER.replace("🌱 Efficiency |", "🌱 Efficiency | 💰 Value |"),
"|" + " --- |" * 8,
f"| 95 % | 2 % | {efficiency} % | {value} | – | 100 % | {score} | {verdict} |",
],
)
self.assertEqual(result.returncode, code)
def test_commit_table(self):
tmp = tempfile.TemporaryDirectory()
self.addCleanup(tmp.cleanup)
@@ -191,6 +253,15 @@ class TestScore(unittest.TestCase):
result = self._run(*flags(BASE), "--commit-table", str(repo), "--commit", "HEAD", cwd=repo)
self.assertEqual(result.stdout.splitlines()[0], "| " + " | ".join([*head, "🔖 Commit"]) + " " + HEADER)
self.assertEqual(result.stdout.splitlines()[2], f"| {repo.resolve()} | main | 2 files +3/-1 | {sha} " + score)
result = self._run(*flags(BASE), "--commit-table", str(repo), "--commit", "HEAD", "--value", "?", cwd=repo)
self.assertEqual(
result.stdout.splitlines()[0],
"| " + " | ".join([*head, "🔖 Commit"]) + " " + HEADER.replace("🌱 Efficiency |", "🌱 Efficiency | 💰 Value |"),
)
self.assertEqual(
result.stdout.splitlines()[2],
f"| {repo.resolve()} | main | 2 files +3/-1 | {sha} " + score.replace("| 100 % |", "| 100 % | ? |", 1),
)
git("checkout", "-q", "--detach", "HEAD")
result = self._run(*flags(BASE), "--commit-table", str(repo), cwd=repo)
self.assertEqual(result.stdout.splitlines()[2], f"| {repo.resolve()} | detached at {sha} | 0 files +0/-0 " + score)
@@ -211,7 +282,17 @@ class TestScore(unittest.TestCase):
with self.subTest(missing=missing):
values = {k: v for k, v in BASE.items() if k != missing}
self.assertEqual(self._run(*flags(values), cwd=REPO_ROOT).returncode, 2)
for extra in (("--commit", "HEAD"), ("--table", "--commit-table", str(REPO_ROOT))):
for extra in (
("--commit", "HEAD"),
("--table", "--commit-table", str(REPO_ROOT)),
("--message-block", "--table"),
("--message-block", "--commit-table", str(REPO_ROOT)),
("--value", "1:20"),
("--table", "--value", "1-20"),
("--table", "--value", "?:?"),
("--table", "--value", "a:b"),
("--table", "--value", ""),
):
with self.subTest(extra=extra):
self.assertEqual(self._run(*flags(BASE), *extra, cwd=REPO_ROOT).returncode, 2)