Rules that can go stale say so — verify_with, expires_when, the sweep, and task_kind='spike' (milestone 312, steps 1–5) #132
@@ -281,3 +281,59 @@ export async function includeAlwaysOnRulebook(projectId: number, rulebookId: num
|
||||
await apiDelete(`/api/projects/${projectId}/exclusions/rulebooks/${rulebookId}`);
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* One row of the staleness sweep. Unlike RuleHeader this carries the CHECK
|
||||
* in full — the reader is about to go and run it, so the text is the point
|
||||
* of the payload rather than the bloat a listing avoids.
|
||||
*/
|
||||
export interface RuleVerificationRow {
|
||||
id: number;
|
||||
title: string;
|
||||
statement: string;
|
||||
tier: RuleTier;
|
||||
topic_id: number | null;
|
||||
project_id: number | null;
|
||||
when_to_apply: string;
|
||||
verify_with: string;
|
||||
expires_when: string;
|
||||
/** A date (YYYY-MM-DD), or the literal "never". */
|
||||
last_verified: string | null;
|
||||
/** Null when never verified — "never" is not zero days ago. */
|
||||
days_since_verified: number | null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Rules asserting a fact that may have gone false, oldest verification
|
||||
* first, never-checked at the top. Rules without a check never appear:
|
||||
* they are decisions, and there is nothing to go and check.
|
||||
*
|
||||
* Not filterable by project — a project reaches rules through project
|
||||
* scope, subscriptions, always-on rulebooks and exclusions, and a filter
|
||||
* missing one of those paths would under-report.
|
||||
*/
|
||||
export async function listRulesDueForVerification(opts: {
|
||||
olderThanDays?: number;
|
||||
tier?: RuleTier;
|
||||
neverOnly?: boolean;
|
||||
} = {}): Promise<{ rules: RuleVerificationRow[]; total: number }> {
|
||||
const q = new URLSearchParams();
|
||||
if (opts.olderThanDays) q.set("older_than_days", String(opts.olderThanDays));
|
||||
if (opts.tier) q.set("tier", opts.tier);
|
||||
if (opts.neverOnly) q.set("never_only", "true");
|
||||
const qs = q.toString();
|
||||
return apiGet(`/api/rules-due-for-verification${qs ? `?${qs}` : ""}`);
|
||||
}
|
||||
|
||||
/**
|
||||
* Record that a rule's check was RUN, and what it said.
|
||||
*
|
||||
* `stillTrue: false` writes nothing on purpose — a rule whose check failed
|
||||
* is not in a recordable state, it is wrong — so it stays at the top of the
|
||||
* sweep until someone corrects or retires it.
|
||||
*/
|
||||
export async function markRuleVerified(
|
||||
id: number, stillTrue = true,
|
||||
): Promise<Rule & { verified: boolean }> {
|
||||
return apiPost(`/api/rules/${id}/verify`, { still_true: stillTrue });
|
||||
}
|
||||
|
||||
@@ -691,6 +691,97 @@ async def unrelate_rules(relation_id: int) -> dict:
|
||||
raise ValueError(f"relation {relation_id} not found")
|
||||
return {"deleted": relation_id}
|
||||
|
||||
# ── The staleness sweep (milestone 312) ────────────────────────────────
|
||||
|
||||
async def rules_due_for_verification(
|
||||
older_than_days: int = 0, tier: str = "", never_only: bool = False,
|
||||
) -> dict:
|
||||
"""Which standing rules assert a FACT that nobody has confirmed lately.
|
||||
|
||||
A rulebook holds two kinds of thing. Most rules are DECISIONS — how the
|
||||
operator wants to work. They have no truth value and cannot rot. A few
|
||||
assert a fact about someone else's software: what a CI runner does, which
|
||||
tools exist, what a setting is currently set to. Those go false silently,
|
||||
with nobody present, and they keep being handed to every session as
|
||||
binding instructions long after they stopped being true.
|
||||
|
||||
This lists the second kind, oldest verification first, never-checked at
|
||||
the top. Each row carries the rule's `verify_with` in full — you are
|
||||
about to go and run it — plus `expires_when`, and `days_since_verified`.
|
||||
|
||||
Reach for it when you are curating the rulebook, when a rule's advice
|
||||
just contradicted what you observed, or periodically. Then, for each row:
|
||||
run the check, and call mark_rule_verified with what you found.
|
||||
|
||||
Rules with no `verify_with` never appear here. That is correct: they are
|
||||
decisions, and there is nothing to go and check. Do not "fix" their
|
||||
absence by giving them checks — the list is only worth reading while
|
||||
everything on it genuinely can go false.
|
||||
|
||||
Args:
|
||||
older_than_days: only rules last verified longer ago than this.
|
||||
Never-checked rules always qualify. 0 = no age filter.
|
||||
tier: "always_on" or "conditional" to narrow. An always-on constraint
|
||||
that has gone false is the expensive kind — it is preloaded into
|
||||
every session, so a wrong one is wrong everywhere at once.
|
||||
never_only: only rules nobody has ever verified.
|
||||
|
||||
NOT filterable by project, deliberately: a project reaches rules through
|
||||
project scope, subscriptions, always-on rulebooks and exclusions, and a
|
||||
filter that missed one of those paths would UNDER-report — which is the
|
||||
exact failure this whole surface exists to prevent. Read the whole list.
|
||||
"""
|
||||
uid = current_user_id()
|
||||
rules = await rulebooks_svc.rules_due_for_verification(
|
||||
uid, older_than_days=older_than_days, tier=tier, never_only=never_only,
|
||||
)
|
||||
return {
|
||||
"rules": [rulebooks_svc.verification_row(r) for r in rules],
|
||||
"total": len(rules),
|
||||
}
|
||||
|
||||
|
||||
async def mark_rule_verified(rule_id: int, still_true: bool = True) -> dict:
|
||||
"""Record that you ran a rule's check — and what it said.
|
||||
|
||||
Call this AFTER actually running the rule's `verify_with`, never on the
|
||||
strength of the rule sounding plausible. A stamp nobody earned is worse
|
||||
than no stamp: it moves the rule to the bottom of the sweep and buys it
|
||||
another long silence.
|
||||
|
||||
`still_true=False` writes NOTHING. A rule whose check failed is not in a
|
||||
special state to be recorded — it is WRONG, and the only honest next
|
||||
moves are to correct it, retire it, or find out why. So it stays at the
|
||||
top of the sweep until someone deals with it, and the response tells you
|
||||
what the rule said would end it.
|
||||
|
||||
Args:
|
||||
rule_id: the rule whose check you ran.
|
||||
still_true: True if the check passed. False if the fact it asserts is
|
||||
no longer true — say so, that is the outcome worth having.
|
||||
"""
|
||||
uid = current_user_id()
|
||||
rule = await rulebooks_svc.mark_rule_verified(rule_id, uid, still_true)
|
||||
if rule is None:
|
||||
raise ValueError(
|
||||
f"rule {rule_id} not found, or carries no verify_with "
|
||||
f"(nothing to verify is not the same as verified)"
|
||||
)
|
||||
data = await rulebooks_svc.rule_detail(uid, rule)
|
||||
if still_true:
|
||||
data["verified"] = True
|
||||
return data
|
||||
data["verified"] = False
|
||||
data["next"] = (
|
||||
"This rule is no longer true and is still binding on every session "
|
||||
"that loads it. Correct it with update_rule, retire it with "
|
||||
"delete_rule, or open a task to work out what replaced it. Its "
|
||||
"verified_at is deliberately untouched, so it stays at the top of "
|
||||
"rules_due_for_verification until one of those happens."
|
||||
)
|
||||
return data
|
||||
|
||||
|
||||
def register(mcp) -> None:
|
||||
for fn in (
|
||||
list_rulebooks, get_rulebook, create_rulebook, update_rulebook, delete_rulebook,
|
||||
@@ -702,5 +793,6 @@ def register(mcp) -> None:
|
||||
suppress_rule_for_project, unsuppress_rule_for_project,
|
||||
suppress_topic_for_project, unsuppress_topic_for_project,
|
||||
exclude_always_on_rulebook, include_always_on_rulebook,
|
||||
rules_due_for_verification, mark_rule_verified,
|
||||
):
|
||||
mcp.tool(name=fn.__name__)(fn)
|
||||
|
||||
@@ -390,3 +390,56 @@ async def create_project_rule(project_id: int):
|
||||
return jsonify(await rulebooks_svc.rule_detail(
|
||||
get_current_user_id(), rule, data.get("system_ids"),
|
||||
)), 201
|
||||
|
||||
|
||||
# ── The staleness sweep (milestone 312) ────────────────────────────────
|
||||
|
||||
@rulebooks_bp.get("/rules-due-for-verification")
|
||||
@login_required
|
||||
async def rules_due_for_verification():
|
||||
"""Rules that carry a check, oldest verification first, never-checked top.
|
||||
|
||||
Query params: older_than_days, tier, never_only. A rule with no
|
||||
`verify_with` never appears — it is a decision, not a fact.
|
||||
"""
|
||||
uid = get_current_user_id()
|
||||
args = request.args
|
||||
try:
|
||||
older = int(args.get("older_than_days", 0) or 0)
|
||||
except ValueError:
|
||||
return jsonify({"error": "older_than_days must be an integer"}), 400
|
||||
try:
|
||||
rules = await rulebooks_svc.rules_due_for_verification(
|
||||
uid,
|
||||
older_than_days=older,
|
||||
tier=args.get("tier", ""),
|
||||
never_only=args.get("never_only", "").lower() in ("1", "true", "yes"),
|
||||
)
|
||||
except ValueError as exc:
|
||||
# An unrecognised tier is a 400, not a silently narrowed result set:
|
||||
# a filter that quietly answers a different question is the failure
|
||||
# this whole surface exists to catch.
|
||||
return jsonify({"error": str(exc)}), 400
|
||||
return jsonify({
|
||||
"rules": [rulebooks_svc.verification_row(r) for r in rules],
|
||||
"total": len(rules),
|
||||
})
|
||||
|
||||
|
||||
@rulebooks_bp.post("/rules/<int:rule_id>/verify")
|
||||
@login_required
|
||||
async def mark_rule_verified(rule_id: int):
|
||||
"""Record that the rule's check was run. Body: {"still_true": bool}.
|
||||
|
||||
`still_true: false` writes nothing — a rule whose check failed is wrong,
|
||||
not in a recordable state — so it stays at the top of the sweep.
|
||||
"""
|
||||
data = await request.get_json() or {}
|
||||
uid = get_current_user_id()
|
||||
still_true = data.get("still_true", True)
|
||||
rule = await rulebooks_svc.mark_rule_verified(rule_id, uid, bool(still_true))
|
||||
if rule is None:
|
||||
return jsonify({"error": "rule not found, or carries no verify_with"}), 404
|
||||
payload = await rulebooks_svc.rule_detail(uid, rule)
|
||||
payload["verified"] = bool(still_true)
|
||||
return jsonify(payload)
|
||||
|
||||
@@ -11,7 +11,7 @@ import logging
|
||||
from collections.abc import Iterable
|
||||
from typing import Optional
|
||||
|
||||
from sqlalchemy import delete as sql_delete, insert, or_, select
|
||||
from sqlalchemy import and_, delete as sql_delete, insert, or_, select
|
||||
|
||||
from scribe.models import async_session
|
||||
from scribe.models.system import System
|
||||
@@ -1361,3 +1361,150 @@ def rules_payload(applicable: dict) -> dict:
|
||||
"suppressed_topics": applicable.get("suppressed_topics", []),
|
||||
"excluded_always_on": applicable.get("excluded_always_on", []),
|
||||
}
|
||||
|
||||
|
||||
# ── The staleness sweep (milestone 312) ────────────────────────────────
|
||||
|
||||
async def rules_due_for_verification(
|
||||
user_id: int,
|
||||
older_than_days: int = 0,
|
||||
tier: str = "",
|
||||
never_only: bool = False,
|
||||
) -> list[Rule]:
|
||||
"""Rules that carry a check, oldest verification first, never-checked top.
|
||||
|
||||
THE QUERY THIS MILESTONE EXISTS FOR. `verify_with` and `expires_when` are
|
||||
storage; this is what turns them into something that gets acted on. The
|
||||
307 audit cost a session and found four broken rules by luck — this makes
|
||||
the same question a list, and staleness measurable by age instead of
|
||||
discoverable by accident.
|
||||
|
||||
Ordered `verified_at` ASC NULLS FIRST: never-checked outranks
|
||||
checked-long-ago, because a rule nobody has ever confirmed is a claim
|
||||
with no evidence behind it at all.
|
||||
|
||||
Rules with no `verify_with` never appear. That is not an omission — they
|
||||
are decisions, there is nothing to go and check, and listing them would
|
||||
dilute the result until nobody reads it.
|
||||
|
||||
Ownership-scoped exactly like list_rules: a rule reached through an owned
|
||||
rulebook, or scoped to an owned project. Rules have no sharing ACL in this
|
||||
schema — no rule_shares, no rulebook_shares — so there is no wider set to
|
||||
consult here, unlike notes and projects.
|
||||
|
||||
Args:
|
||||
user_id: whose rules.
|
||||
older_than_days: only rules last verified longer ago than this.
|
||||
Never-checked rules always qualify — they are the most overdue
|
||||
thing there is. 0 = no age filter.
|
||||
tier: "always_on" or "conditional" to narrow. Raises on anything else
|
||||
rather than falling back: _valid_tier's silent always_on default
|
||||
is right for a WRITE (the safe direction is to keep binding), and
|
||||
wrong for a FILTER, where it would quietly answer a different
|
||||
question than the one asked.
|
||||
never_only: only rules that have never been verified.
|
||||
"""
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
from scribe.models.project import Project
|
||||
|
||||
if tier and tier not in TIERS:
|
||||
raise ValueError(f"tier must be one of {TIERS}, got {tier!r}")
|
||||
|
||||
async with async_session() as session:
|
||||
stmt = (
|
||||
select(Rule)
|
||||
.outerjoin(RulebookTopic, Rule.topic_id == RulebookTopic.id)
|
||||
.outerjoin(Rulebook, RulebookTopic.rulebook_id == Rulebook.id)
|
||||
.outerjoin(Project, Rule.project_id == Project.id)
|
||||
.where(
|
||||
Rule.deleted_at.is_(None),
|
||||
Rule.verify_with.is_not(None),
|
||||
# One statement rather than two queries merged in Python, so
|
||||
# the ordering below is the database's and cannot disagree
|
||||
# with itself across the two halves of the XOR.
|
||||
or_(
|
||||
and_(
|
||||
Rulebook.owner_user_id == user_id,
|
||||
Rulebook.deleted_at.is_(None),
|
||||
RulebookTopic.deleted_at.is_(None),
|
||||
),
|
||||
Project.user_id == user_id,
|
||||
),
|
||||
)
|
||||
)
|
||||
if tier:
|
||||
stmt = stmt.where(Rule.tier == tier)
|
||||
if never_only:
|
||||
stmt = stmt.where(Rule.verified_at.is_(None))
|
||||
elif older_than_days > 0:
|
||||
cutoff = datetime.now(timezone.utc) - timedelta(days=older_than_days)
|
||||
stmt = stmt.where(
|
||||
or_(Rule.verified_at.is_(None), Rule.verified_at < cutoff)
|
||||
)
|
||||
stmt = stmt.order_by(Rule.verified_at.asc().nullsfirst(), Rule.id)
|
||||
return list((await session.execute(stmt)).scalars().all())
|
||||
|
||||
|
||||
def verification_row(rule: Rule) -> dict:
|
||||
"""One row of the sweep — the CHECK in full, unlike rule_brief.
|
||||
|
||||
The opposite call from a listing: here the caller is about to go and run
|
||||
the check, so the text they need is the point of the payload rather than
|
||||
the bloat. `days_since` is computed rather than left to the reader,
|
||||
because "2026-06-14" and "74 days" prompt different reactions and only
|
||||
one of them is the question being asked.
|
||||
"""
|
||||
from datetime import datetime, timezone
|
||||
|
||||
days = None
|
||||
if rule.verified_at is not None:
|
||||
stamp = rule.verified_at
|
||||
if stamp.tzinfo is None:
|
||||
stamp = stamp.replace(tzinfo=timezone.utc)
|
||||
days = (datetime.now(timezone.utc) - stamp).days
|
||||
return {
|
||||
"id": rule.id,
|
||||
"title": rule.title,
|
||||
"statement": rule.statement,
|
||||
"tier": rule.tier,
|
||||
"topic_id": rule.topic_id,
|
||||
"project_id": rule.project_id,
|
||||
"when_to_apply": rule.when_to_apply or "",
|
||||
"verify_with": rule.verify_with or "",
|
||||
"expires_when": rule.expires_when or "",
|
||||
"last_verified": last_verified_label(rule),
|
||||
"days_since_verified": days,
|
||||
}
|
||||
|
||||
|
||||
async def mark_rule_verified(
|
||||
rule_id: int, user_id: int, still_true: bool = True,
|
||||
) -> Optional[Rule]:
|
||||
"""Stamp a rule as verified — or, when the check FAILED, refuse to.
|
||||
|
||||
A failing check is the outcome worth having, and the asymmetry is
|
||||
deliberate: passing writes a stamp, failing writes nothing. There is no
|
||||
"verified false" state to record, because a rule whose check failed is
|
||||
not a rule in a special condition — it is a rule that is WRONG, and the
|
||||
only honest resolutions are to correct it, retire it, or find out why.
|
||||
Recording the failure as a flag would let it sit there being false with
|
||||
the sweep quietly satisfied that someone had looked.
|
||||
|
||||
So a failed check leaves `verified_at` untouched, and the rule stays at
|
||||
the top of the sweep until someone actually deals with it.
|
||||
|
||||
Returns None when the rule is not found, not owned, or carries no
|
||||
`verify_with` — nothing to verify is a different answer from verified.
|
||||
"""
|
||||
from datetime import datetime, timezone
|
||||
|
||||
async with async_session() as session:
|
||||
rule = await _fetch_owned_rule(session, rule_id, user_id)
|
||||
if rule is None or not rule.verify_with:
|
||||
return None
|
||||
if still_true:
|
||||
rule.verified_at = datetime.now(timezone.utc)
|
||||
await session.commit()
|
||||
await session.refresh(rule)
|
||||
return rule
|
||||
|
||||
@@ -134,3 +134,142 @@ async def test_editing_anything_else_leaves_the_stamp_alone(constraint):
|
||||
row = await _row(constraint["rule"])
|
||||
assert row.verified_at is not None
|
||||
assert row.why.startswith("act_runner picks the shell")
|
||||
|
||||
|
||||
# ── the sweep itself (step 3) ──────────────────────────────────────────
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def rulebook_of_three():
|
||||
"""A decision, a never-checked constraint, and a long-ago-checked one.
|
||||
|
||||
Three rows because the sweep's whole value is an ORDER, and an order
|
||||
cannot be asserted with fewer.
|
||||
"""
|
||||
async with async_session() as s:
|
||||
owner = await ensure_user(s, "sweep_owner")
|
||||
uid = owner.id
|
||||
await s.commit()
|
||||
|
||||
book = await rulebooks_svc.create_rulebook(uid, "Sweep fixture")
|
||||
topic = await rulebooks_svc.create_topic(book.id, uid, "mixed")
|
||||
decision = await rulebooks_svc.create_rule(
|
||||
topic.id, uid, "dev is home", "Work directly on dev.",
|
||||
)
|
||||
never = await rulebooks_svc.create_rule(
|
||||
topic.id, uid, "The runner has no bash", "Use POSIX sh.",
|
||||
verify_with="read the workflow's shell setting",
|
||||
)
|
||||
stale = await rulebooks_svc.create_rule(
|
||||
topic.id, uid, "Bumps need a dashboard tick", "Tick it first.",
|
||||
verify_with="cat CI-runner/renovate/config.js",
|
||||
tier="conditional",
|
||||
)
|
||||
async with async_session() as s:
|
||||
row = await s.get(Rule, stale.id)
|
||||
row.verified_at = datetime(2026, 1, 1, tzinfo=timezone.utc)
|
||||
await s.commit()
|
||||
return {
|
||||
"uid": uid, "decision": decision.id,
|
||||
"never": never.id, "stale": stale.id,
|
||||
}
|
||||
|
||||
|
||||
async def test_a_rule_with_no_check_is_never_in_the_sweep(rulebook_of_three):
|
||||
"""The common case, and the one that keeps the list worth reading.
|
||||
|
||||
Most rules are decisions. If they appeared here the sweep would be the
|
||||
rulebook, and nobody would read it twice.
|
||||
"""
|
||||
rules = await rulebooks_svc.rules_due_for_verification(rulebook_of_three["uid"])
|
||||
assert rulebook_of_three["decision"] not in [r.id for r in rules]
|
||||
|
||||
|
||||
async def test_never_checked_outranks_checked_long_ago(rulebook_of_three):
|
||||
"""NULLS FIRST is the ordering decision this surface turns on.
|
||||
|
||||
Postgres sorts NULLs LAST by default on an ASC ordering, which would put
|
||||
the rules nobody has ever confirmed at the BOTTOM — behind every rule
|
||||
that at least once had someone look at it. That is exactly backwards: a
|
||||
claim with no evidence at all outranks an old one.
|
||||
"""
|
||||
ids = [r.id for r in await rulebooks_svc.rules_due_for_verification(
|
||||
rulebook_of_three["uid"]
|
||||
)]
|
||||
assert ids.index(rulebook_of_three["never"]) < ids.index(rulebook_of_three["stale"])
|
||||
|
||||
|
||||
async def test_verifying_a_rule_moves_it_off_the_top(rulebook_of_three):
|
||||
"""The loop closing: check it, stamp it, and it stops being the question."""
|
||||
await rulebooks_svc.mark_rule_verified(
|
||||
rulebook_of_three["never"], rulebook_of_three["uid"], still_true=True,
|
||||
)
|
||||
ids = [r.id for r in await rulebooks_svc.rules_due_for_verification(
|
||||
rulebook_of_three["uid"]
|
||||
)]
|
||||
# Still present — verified is not retired, and it will come due again.
|
||||
assert rulebook_of_three["never"] in ids
|
||||
assert ids.index(rulebook_of_three["stale"]) < ids.index(rulebook_of_three["never"])
|
||||
|
||||
|
||||
async def test_a_failed_check_writes_nothing(rulebook_of_three):
|
||||
"""The asymmetry that keeps the sweep honest.
|
||||
|
||||
There is no "verified false" state, because a rule whose check failed is
|
||||
not in a special condition — it is WRONG. Recording the failure would let
|
||||
it sit there being false with the sweep satisfied that someone looked.
|
||||
"""
|
||||
before = await _row(rulebook_of_three["stale"])
|
||||
await rulebooks_svc.mark_rule_verified(
|
||||
rulebook_of_three["stale"], rulebook_of_three["uid"], still_true=False,
|
||||
)
|
||||
after = await _row(rulebook_of_three["stale"])
|
||||
assert after.verified_at == before.verified_at
|
||||
|
||||
|
||||
async def test_a_rule_with_no_check_cannot_be_verified(rulebook_of_three):
|
||||
"""Nothing to verify is a different answer from verified — and stamping
|
||||
one would put a decision into a sweep it has no business being in."""
|
||||
assert await rulebooks_svc.mark_rule_verified(
|
||||
rulebook_of_three["decision"], rulebook_of_three["uid"],
|
||||
) is None
|
||||
|
||||
|
||||
async def test_never_only_and_the_age_filter_narrow_to_what_they_say(rulebook_of_three):
|
||||
uid = rulebook_of_three["uid"]
|
||||
# Membership, not equality: the integration lane shares one database for
|
||||
# the whole run and this fixture is function-scoped, so this owner has
|
||||
# accumulated rules from earlier tests. Asserting the exact list would
|
||||
# pass alone and fail in the suite.
|
||||
never_ids = [r.id for r in await rulebooks_svc.rules_due_for_verification(
|
||||
uid, never_only=True,
|
||||
)]
|
||||
assert rulebook_of_three["never"] in never_ids
|
||||
assert rulebook_of_three["stale"] not in never_ids
|
||||
assert rulebook_of_three["decision"] not in never_ids
|
||||
|
||||
# A rule checked in January is well past any sane window; one never
|
||||
# checked always qualifies, because it is the most overdue thing there is.
|
||||
aged = [r.id for r in await rulebooks_svc.rules_due_for_verification(
|
||||
uid, older_than_days=30,
|
||||
)]
|
||||
assert rulebook_of_three["stale"] in aged
|
||||
assert rulebook_of_three["never"] in aged
|
||||
|
||||
|
||||
async def test_the_tier_filter_narrows_to_one_tier(rulebook_of_three):
|
||||
ids = [r.id for r in await rulebooks_svc.rules_due_for_verification(
|
||||
rulebook_of_three["uid"], tier="conditional",
|
||||
)]
|
||||
assert rulebook_of_three["stale"] in ids
|
||||
assert rulebook_of_three["never"] not in ids
|
||||
|
||||
|
||||
async def test_another_users_rules_are_not_in_your_sweep(rulebook_of_three):
|
||||
"""Rules are ownership-scoped: there is no rule-sharing ACL in this
|
||||
schema, so the only correct answer is your own rules."""
|
||||
async with async_session() as s:
|
||||
stranger = await ensure_user(s, "sweep_stranger")
|
||||
sid = stranger.id
|
||||
await s.commit()
|
||||
|
||||
assert await rulebooks_svc.rules_due_for_verification(sid) == []
|
||||
|
||||
@@ -217,13 +217,20 @@ async def test_unsubscribe_project_from_rulebook_calls_service():
|
||||
assert mock.called
|
||||
|
||||
|
||||
def test_register_attaches_all_sixteen_tools():
|
||||
"""register(mcp) should call mcp.tool(name=...) for all 16 tools."""
|
||||
def test_register_attaches_every_tool():
|
||||
"""Every tool in the module reaches the server.
|
||||
|
||||
The count is the guard: a function added to the module but left out of
|
||||
register()'s tuple is invisible to callers and raises nothing. The name
|
||||
said "sixteen" for ten tools' worth of growth — the number lives in the
|
||||
assertion, not the title, so it cannot drift again.
|
||||
"""
|
||||
from scribe.mcp.tools.rulebooks import register
|
||||
mcp = FakeMCP()
|
||||
|
||||
register(mcp)
|
||||
assert len(mcp.names) == 26 # +relate_rules/unrelate_rules (milestone 307)
|
||||
# 26 through milestone 307, +2 for the staleness sweep (milestone 312).
|
||||
assert len(mcp.names) == 28
|
||||
# spot-check a few names
|
||||
assert "list_rulebooks" in mcp.names
|
||||
assert "create_rule" in mcp.names
|
||||
@@ -234,6 +241,9 @@ def test_register_attaches_all_sixteen_tools():
|
||||
assert "include_always_on_rulebook" in mcp.names
|
||||
assert "create_project_rule" in mcp.names
|
||||
assert "suppress_rule_for_project" in mcp.names
|
||||
# milestone 312: the sweep, and the stamp that answers it
|
||||
assert "rules_due_for_verification" in mcp.names
|
||||
assert "mark_rule_verified" in mcp.names
|
||||
assert "unsuppress_rule_for_project" in mcp.names
|
||||
assert "suppress_topic_for_project" in mcp.names
|
||||
assert "unsuppress_topic_for_project" in mcp.names
|
||||
|
||||
@@ -446,3 +446,62 @@ def test_the_check_text_itself_never_enters_a_listing():
|
||||
))
|
||||
assert "verify_with" not in out
|
||||
assert "expires_when" not in out
|
||||
|
||||
|
||||
# ── the sweep's row shape (milestone 312 step 3) ────────────────────────
|
||||
|
||||
def test_a_sweep_row_carries_the_check_in_full():
|
||||
"""The OPPOSITE call from rule_brief, and deliberately so.
|
||||
|
||||
A listing omits the depth because nobody reading it wants to act on one
|
||||
rule. A sweep row exists to be acted on — the reader is about to go and
|
||||
run the check — so the text is the payload's point, not its bloat.
|
||||
"""
|
||||
from scribe.services.rulebooks import verification_row
|
||||
|
||||
row = verification_row(fake_rule(
|
||||
verify_with="cat CI-runner/renovate/config.js",
|
||||
expires_when="dependencyDashboardApproval is turned off",
|
||||
when_to_apply="when a dependency bump is in play",
|
||||
))
|
||||
assert row["verify_with"] == "cat CI-runner/renovate/config.js"
|
||||
assert row["expires_when"] == "dependencyDashboardApproval is turned off"
|
||||
assert row["when_to_apply"] == "when a dependency bump is in play"
|
||||
assert row["tier"] == "always_on"
|
||||
|
||||
|
||||
def test_never_verified_reports_no_day_count_rather_than_zero():
|
||||
""""Never" is not "0 days ago" — the second reads as freshly checked.
|
||||
|
||||
Getting this wrong would invert the row's meaning for exactly the rules
|
||||
that most need attention.
|
||||
"""
|
||||
from scribe.services.rulebooks import verification_row
|
||||
|
||||
row = verification_row(fake_rule(verify_with="read the workflow"))
|
||||
assert row["last_verified"] == "never"
|
||||
assert row["days_since_verified"] is None
|
||||
|
||||
|
||||
def test_a_verified_row_counts_the_days():
|
||||
from datetime import timedelta
|
||||
|
||||
from scribe.services.rulebooks import verification_row
|
||||
|
||||
row = verification_row(fake_rule(
|
||||
verify_with="read the workflow",
|
||||
verified_at=datetime.now(timezone.utc) - timedelta(days=74, hours=1),
|
||||
))
|
||||
assert row["days_since_verified"] == 74
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_an_unrecognised_tier_filter_raises_rather_than_narrowing():
|
||||
"""_valid_tier's silent always_on fallback is right for a WRITE — a typo
|
||||
should leave a rule binding. It is wrong for a FILTER, where the same
|
||||
fallback would quietly answer a different question than the one asked and
|
||||
return a short list that looks like good news."""
|
||||
from scribe.services.rulebooks import rules_due_for_verification
|
||||
|
||||
with pytest.raises(ValueError, match="tier must be one of"):
|
||||
await rules_due_for_verification(7, tier="occasionally")
|
||||
|
||||
Reference in New Issue
Block a user