bvandeusenandClaude Opus 5
410d616c22
CI & Build / Python lint (push) Successful in 4s
CI & Build / Plugin hooks (push) Successful in 10s
CI & Build / TypeScript typecheck (push) Successful in 34s
CI & Build / integration (push) Successful in 32s
CI & Build / Python tests (push) Successful in 1m8s
CI & Build / Build & push image (push) Successful in 35s
The query the last two steps were storage for. `rules_due_for_verification` returns every rule carrying a `verify_with`, ordered by `verified_at` ASC NULLS FIRST, each row carrying the check IN FULL — the opposite call from rule_brief, because the reader is about to go and run it. NULLS FIRST is the ordering this turns on. Postgres sorts NULLs last on an ASC ordering, which would put the rules nobody has ever confirmed BEHIND every rule someone once looked at. Exactly backwards: a claim with no evidence at all outranks an old one. Rules with no check never appear, and that is the property that keeps the list worth reading. Most rules are decisions — no truth value, nothing to go and check. If they appeared here the sweep would be the rulebook. `mark_rule_verified(rule_id, still_true)` closes the loop, asymmetrically: passing writes a stamp, FAILING WRITES NOTHING. There is no "verified false" state because a rule whose check failed is not in a special condition, it is wrong — and recording the failure as a flag would let it sit there being false with the sweep satisfied that someone had looked. So it stays at the top until someone corrects or retires it, and the response says so. An unrecognised `tier` filter raises rather than falling back. _valid_tier's silent always_on default is right for a WRITE — a typo should leave a rule binding — and wrong for a FILTER, where the same fallback quietly answers a different question and returns a short list that reads as good news. Deliberately 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 — the exact failure this surface exists to prevent. Said so in the docstring rather than shipping a half-correct filter. Ownership-scoped like every other rule read (owned rulebook, or owned project), in ONE statement with an OR across the XOR rather than two queries merged in Python, so the ordering is the database's and cannot disagree with itself. Note that rules have no sharing ACL in this schema — no rule_shares, no rulebook_shares — so there is no wider set for access.py to consult here. Also fixes a test title that had been lying for ten tools: "all sixteen tools" asserted 26. The number now lives only in the assertion. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
1511 lines
60 KiB
Python
1511 lines
60 KiB
Python
"""Rulebook / topic / rule service layer — single source of truth used by
|
|
both routes/rulebooks.py and mcp/tools/rulebooks.py.
|
|
|
|
Ownership enforcement: every function takes user_id and scopes through
|
|
rulebooks.owner_user_id. Functions return models or model.to_dict() output
|
|
depending on the caller's needs (mirroring services/events.py pattern).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from collections.abc import Iterable
|
|
from typing import Optional
|
|
|
|
from sqlalchemy import and_, delete as sql_delete, insert, or_, select
|
|
|
|
from scribe.models import async_session
|
|
from scribe.models.system import System
|
|
from scribe.models.rulebook import Rulebook
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
# ── Rulebook CRUD ────────────────────────────────────────────────────────
|
|
|
|
async def create_rulebook(
|
|
user_id: int, title: str, description: str = "",
|
|
) -> Rulebook:
|
|
"""Create a new rulebook owned by user_id. Returns the persisted model."""
|
|
async with async_session() as session:
|
|
rb = Rulebook(
|
|
owner_user_id=user_id, title=title, description=description or None,
|
|
)
|
|
session.add(rb)
|
|
await session.commit()
|
|
await session.refresh(rb)
|
|
return rb
|
|
|
|
|
|
async def list_rulebooks(user_id: int) -> list[Rulebook]:
|
|
"""List rulebooks owned by user_id, ordered by title."""
|
|
async with async_session() as session:
|
|
result = await session.execute(
|
|
select(Rulebook)
|
|
.where(Rulebook.owner_user_id == user_id, Rulebook.deleted_at.is_(None))
|
|
.order_by(Rulebook.title)
|
|
)
|
|
return list(result.scalars().all())
|
|
|
|
|
|
async def get_rulebook(rulebook_id: int, user_id: int) -> Optional[Rulebook]:
|
|
"""Get a rulebook by id, scoped to user_id. None if not owned or not found."""
|
|
async with async_session() as session:
|
|
result = await session.execute(
|
|
select(Rulebook).where(
|
|
Rulebook.id == rulebook_id,
|
|
Rulebook.owner_user_id == user_id,
|
|
Rulebook.deleted_at.is_(None),
|
|
)
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def update_rulebook(
|
|
rulebook_id: int, user_id: int, **fields,
|
|
) -> Optional[Rulebook]:
|
|
"""Partial update. Returns updated rulebook or None if not found."""
|
|
async with async_session() as session:
|
|
result = await session.execute(
|
|
select(Rulebook).where(
|
|
Rulebook.id == rulebook_id,
|
|
Rulebook.owner_user_id == user_id,
|
|
)
|
|
)
|
|
rb = result.scalar_one_or_none()
|
|
if rb is None:
|
|
return None
|
|
allowed = {"title", "description", "always_on"}
|
|
for key, value in fields.items():
|
|
if key in allowed and value is not None:
|
|
setattr(rb, key, value)
|
|
await session.commit()
|
|
await session.refresh(rb)
|
|
return rb
|
|
|
|
|
|
async def delete_rulebook(rulebook_id: int, user_id: int) -> None:
|
|
"""Delete a rulebook. Cascade-deletes topics, rules, subscriptions."""
|
|
async with async_session() as session:
|
|
result = await session.execute(
|
|
select(Rulebook).where(
|
|
Rulebook.id == rulebook_id,
|
|
Rulebook.owner_user_id == user_id,
|
|
)
|
|
)
|
|
rb = result.scalar_one_or_none()
|
|
if rb is None:
|
|
return
|
|
await session.delete(rb)
|
|
await session.commit()
|
|
|
|
|
|
async def find_rulebook_by_title(
|
|
user_id: int, title: str,
|
|
) -> Optional[Rulebook]:
|
|
"""Used by the port script for the dupe-guard. None if not found."""
|
|
async with async_session() as session:
|
|
result = await session.execute(
|
|
select(Rulebook).where(
|
|
Rulebook.owner_user_id == user_id,
|
|
Rulebook.title == title,
|
|
Rulebook.deleted_at.is_(None),
|
|
)
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
# ── Topic CRUD ──────────────────────────────────────────────────────────
|
|
|
|
from scribe.models.rulebook import RulebookTopic
|
|
|
|
|
|
async def _assert_rulebook_owned(session, rulebook_id: int, user_id: int) -> None:
|
|
"""Raise ValueError if rulebook doesn't exist or isn't owned by user.
|
|
Centralizes ownership check used by all topic/rule operations.
|
|
"""
|
|
result = await session.execute(
|
|
select(Rulebook).where(
|
|
Rulebook.id == rulebook_id,
|
|
Rulebook.owner_user_id == user_id,
|
|
Rulebook.deleted_at.is_(None),
|
|
)
|
|
)
|
|
if result.scalar_one_or_none() is None:
|
|
raise ValueError(f"rulebook {rulebook_id} not found")
|
|
|
|
|
|
async def create_topic(
|
|
rulebook_id: int, user_id: int, title: str,
|
|
description: str = "", order_index: int = 0,
|
|
) -> RulebookTopic:
|
|
async with async_session() as session:
|
|
await _assert_rulebook_owned(session, rulebook_id, user_id)
|
|
topic = RulebookTopic(
|
|
rulebook_id=rulebook_id,
|
|
title=title,
|
|
description=description or None,
|
|
order_index=order_index,
|
|
)
|
|
session.add(topic)
|
|
await session.commit()
|
|
await session.refresh(topic)
|
|
return topic
|
|
|
|
|
|
async def list_topics(rulebook_id: int, user_id: int) -> list[RulebookTopic]:
|
|
async with async_session() as session:
|
|
await _assert_rulebook_owned(session, rulebook_id, user_id)
|
|
result = await session.execute(
|
|
select(RulebookTopic)
|
|
.where(
|
|
RulebookTopic.rulebook_id == rulebook_id,
|
|
RulebookTopic.deleted_at.is_(None),
|
|
)
|
|
.order_by(RulebookTopic.order_index, RulebookTopic.title)
|
|
)
|
|
return list(result.scalars().all())
|
|
|
|
|
|
async def get_topic(topic_id: int, user_id: int) -> Optional[RulebookTopic]:
|
|
"""Get a topic, scoped via the rulebook owner."""
|
|
async with async_session() as session:
|
|
result = await session.execute(
|
|
select(RulebookTopic)
|
|
.join(Rulebook, RulebookTopic.rulebook_id == Rulebook.id)
|
|
.where(
|
|
RulebookTopic.id == topic_id,
|
|
Rulebook.owner_user_id == user_id,
|
|
RulebookTopic.deleted_at.is_(None),
|
|
Rulebook.deleted_at.is_(None),
|
|
)
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def update_topic(
|
|
topic_id: int, user_id: int, **fields,
|
|
) -> Optional[RulebookTopic]:
|
|
async with async_session() as session:
|
|
result = await session.execute(
|
|
select(RulebookTopic)
|
|
.join(Rulebook, RulebookTopic.rulebook_id == Rulebook.id)
|
|
.where(
|
|
RulebookTopic.id == topic_id,
|
|
Rulebook.owner_user_id == user_id,
|
|
)
|
|
)
|
|
topic = result.scalar_one_or_none()
|
|
if topic is None:
|
|
return None
|
|
allowed = {"title", "description", "order_index"}
|
|
for key, value in fields.items():
|
|
if key in allowed and value is not None:
|
|
setattr(topic, key, value)
|
|
await session.commit()
|
|
await session.refresh(topic)
|
|
return topic
|
|
|
|
|
|
async def delete_topic(topic_id: int, user_id: int) -> None:
|
|
async with async_session() as session:
|
|
result = await session.execute(
|
|
select(RulebookTopic)
|
|
.join(Rulebook, RulebookTopic.rulebook_id == Rulebook.id)
|
|
.where(
|
|
RulebookTopic.id == topic_id,
|
|
Rulebook.owner_user_id == user_id,
|
|
)
|
|
)
|
|
topic = result.scalar_one_or_none()
|
|
if topic is None:
|
|
return
|
|
await session.delete(topic)
|
|
await session.commit()
|
|
|
|
|
|
# ── Rule CRUD ──────────────────────────────────────────────────────────
|
|
|
|
from scribe.models.rulebook import Rule, RuleRelation, rule_systems
|
|
|
|
|
|
async def _assert_topic_owned(session, topic_id: int, user_id: int) -> None:
|
|
"""Raise ValueError if topic doesn't exist or isn't in user's rulebook."""
|
|
result = await session.execute(
|
|
select(RulebookTopic)
|
|
.join(Rulebook, RulebookTopic.rulebook_id == Rulebook.id)
|
|
.where(
|
|
RulebookTopic.id == topic_id,
|
|
Rulebook.owner_user_id == user_id,
|
|
RulebookTopic.deleted_at.is_(None),
|
|
Rulebook.deleted_at.is_(None),
|
|
)
|
|
)
|
|
if result.scalar_one_or_none() is None:
|
|
raise ValueError(f"topic {topic_id} not found")
|
|
|
|
|
|
async def _assert_project_owned(session, project_id: int, user_id: int) -> None:
|
|
"""Raise ValueError if project doesn't exist or isn't owned by user."""
|
|
from scribe.models.project import Project
|
|
result = await session.execute(
|
|
select(Project).where(
|
|
Project.id == project_id,
|
|
Project.user_id == user_id,
|
|
Project.deleted_at.is_(None),
|
|
)
|
|
)
|
|
if result.scalar_one_or_none() is None:
|
|
raise ValueError(f"project {project_id} not found")
|
|
|
|
|
|
async def _assert_rulebook_rule_owned(session, rule_id: int, user_id: int) -> None:
|
|
"""Raise ValueError if rule isn't a rulebook rule the user owns.
|
|
|
|
Project-scoped rules (Rule.project_id set, topic_id NULL) are NOT
|
|
suppressible — they belong to the project; delete them instead. This
|
|
helper deliberately excludes them.
|
|
"""
|
|
from scribe.models.rulebook import Rule
|
|
result = await session.execute(
|
|
select(Rule)
|
|
.join(RulebookTopic, Rule.topic_id == RulebookTopic.id)
|
|
.join(Rulebook, RulebookTopic.rulebook_id == Rulebook.id)
|
|
.where(
|
|
Rule.id == rule_id,
|
|
Rulebook.owner_user_id == user_id,
|
|
Rule.deleted_at.is_(None),
|
|
RulebookTopic.deleted_at.is_(None),
|
|
Rulebook.deleted_at.is_(None),
|
|
)
|
|
)
|
|
if result.scalar_one_or_none() is None:
|
|
raise ValueError(f"rule {rule_id} not found or not a rulebook rule")
|
|
|
|
|
|
# The vocabularies migration 0088's CHECK constraints enforce. Named here so
|
|
# a caller can be corrected before the database refuses it (rule 36 keeps the
|
|
# two in step; this keeps the error readable).
|
|
TIERS = ("always_on", "conditional")
|
|
RELATION_KINDS = ("co_surfaces", "overrides", "elaborates")
|
|
|
|
|
|
# The rule columns that are nullable, and therefore the ones where EMPTY has
|
|
# to mean empty. A write that stores "" leaves a column that is not NULL and
|
|
# not content — `verify_with IS NOT NULL` would then be true for a rule with
|
|
# no check, and the staleness sweep would list rules it should never see.
|
|
# Normalising here, at the one service seam, is what makes "unset" a single
|
|
# state instead of two that read alike through to_dict's `or ""`.
|
|
NULLABLE_RULE_TEXT = (
|
|
"why", "how_to_apply", "when_to_apply", "verify_with", "expires_when",
|
|
)
|
|
|
|
|
|
def _valid_tier(tier: str) -> str:
|
|
"""An unrecognised tier falls back to always_on — the SAFE direction.
|
|
|
|
Getting this wrong the other way would silently stop a rule binding, which
|
|
is the one failure this whole milestone exists to prevent. A rule that
|
|
preloads when it did not need to costs context; a rule that quietly stops
|
|
preloading costs the behaviour it was written for.
|
|
"""
|
|
return tier if tier in TIERS else "always_on"
|
|
|
|
|
|
def last_verified_label(rule: Rule) -> str | None:
|
|
"""How long ago the rule's check passed — None when it carries no check.
|
|
|
|
One helper because two surfaces need the same answer and the brief-dict
|
|
lesson in rule_brief's docstring is what happens otherwise: three copies
|
|
that had already drifted. `None` means "this rule is a decision, the
|
|
question does not apply"; "never" means "it is a fact and nobody has
|
|
confirmed it" — a distinction worth keeping, because the second is the
|
|
one worth acting on.
|
|
"""
|
|
if not rule.verify_with:
|
|
return None
|
|
return rule.verified_at.date().isoformat() if rule.verified_at else "never"
|
|
|
|
|
|
def rule_brief(rule: Rule, **extra) -> dict:
|
|
"""The shape a rule takes when it is SURFACED rather than opened.
|
|
|
|
One builder for every payload that hands rules to an agent, because there
|
|
were three copies of this dict and they had already diverged — two carried
|
|
`topic_id`, one didn't, and none carried the timestamps the model has held
|
|
all along. That omission is why a rule written before the capability it
|
|
duplicates was indistinguishable, at read time, from one still doing work
|
|
(the FabledCurator case, note 3026).
|
|
|
|
`updated_at` is a DATE, not a stamp: the question it answers is "how old
|
|
is this?", and a full ISO string across an always-on set is ~2k characters
|
|
of payload for a precision nobody reads.
|
|
|
|
`why` and `how_to_apply` are deliberately NOT here — they are the depth a
|
|
caller gets from get_rule, and putting them in every listing is the bloat
|
|
this milestone is about.
|
|
"""
|
|
out = {
|
|
"id": rule.id,
|
|
"title": rule.title,
|
|
"statement": rule.statement,
|
|
"topic_id": rule.topic_id,
|
|
"tier": rule.tier,
|
|
"updated_at": rule.updated_at.date().isoformat() if rule.updated_at else None,
|
|
}
|
|
# Attached only when present (#2483: never a null key that reads as a
|
|
# capability the record doesn't have).
|
|
if rule.when_to_apply:
|
|
out["when_to_apply"] = rule.when_to_apply
|
|
if rule.arose_from_id:
|
|
out["arose_from_id"] = rule.arose_from_id
|
|
# Present ONLY on a rule that carries a check — its presence is the
|
|
# signal, and it says two things at once: this rule asserts a fact that
|
|
# can go false, and here is how long ago anyone confirmed it. The check
|
|
# text itself stays in get_rule; a listing needs to know WHICH rules can
|
|
# rot, not how to test them. "never" rather than null, per #2483: a key
|
|
# that reads as an unused capability is a different claim from a rule
|
|
# nobody has ever verified.
|
|
stamp = last_verified_label(rule)
|
|
if stamp:
|
|
out["last_verified"] = stamp
|
|
out.update({k: v for k, v in extra.items() if v is not None})
|
|
return out
|
|
|
|
|
|
def _refresh_rule_embedding(rule: Rule) -> None:
|
|
"""Re-index a rule after a write. Fire-and-forget, like the note twin.
|
|
|
|
Lazy import so this module doesn't pull in the embedder; every exception
|
|
swallowed because a rule that SAVED must not fail on its index refresh —
|
|
a stale vector costs a missed search hit, a raised exception costs the
|
|
write. No running loop (unit tests, scripts) is ordinary, not an error.
|
|
"""
|
|
try:
|
|
import asyncio
|
|
|
|
from scribe.services.embeddings import upsert_rule_embedding
|
|
|
|
asyncio.create_task(
|
|
upsert_rule_embedding(
|
|
rule.id, rule.title, rule.statement, rule.when_to_apply,
|
|
)
|
|
)
|
|
except RuntimeError:
|
|
pass # no running loop — a sync caller, not a failure
|
|
except Exception: # noqa: BLE001 - never let indexing break a write
|
|
logger.exception("embedding refresh failed for rule %s", rule.id)
|
|
|
|
|
|
async def co_surfaced_partners(
|
|
user_id: int, rule_ids: list[int], exclude_ids: set[int] | None = None,
|
|
) -> list[Rule]:
|
|
"""Rules that must arrive WITH the given ones, because they fail together.
|
|
|
|
This is the whole reason `co_surfaces` exists. Rule 144 was split off rule
|
|
46 and folded back into it the same day, on the correct observation that
|
|
"either rule could surface without the other and miss exposing a project to
|
|
what the entire shape is intended to be." Merging was the only fix
|
|
available; this is the fix that should have been available.
|
|
|
|
Two limits, both deliberate:
|
|
|
|
- Only rules the caller OWNS. An edge is not a back door into someone
|
|
else's rulebook.
|
|
- `exclude_ids` is honoured, and callers pass the project's SUPPRESSIONS.
|
|
A project that explicitly muted a rule should not have it dragged back in
|
|
by an edge — the suppression is a decision, and the edge does not
|
|
outrank it.
|
|
"""
|
|
if not rule_ids:
|
|
return []
|
|
known = set(rule_ids) | (exclude_ids or set())
|
|
async with async_session() as session:
|
|
edges = (await session.execute(
|
|
select(RuleRelation).where(
|
|
RuleRelation.kind == "co_surfaces",
|
|
or_(
|
|
RuleRelation.from_rule_id.in_(rule_ids),
|
|
RuleRelation.to_rule_id.in_(rule_ids),
|
|
),
|
|
)
|
|
)).scalars().all()
|
|
partners = {
|
|
(edge.to_rule_id if edge.from_rule_id in known else edge.from_rule_id)
|
|
for edge in edges
|
|
} - known
|
|
if not partners:
|
|
return []
|
|
# Ownership re-checked per partner rather than assumed from the edge.
|
|
out = []
|
|
for partner_id in sorted(partners):
|
|
rule = await _fetch_owned_rule(session, partner_id, user_id)
|
|
if rule is not None:
|
|
out.append(rule)
|
|
return out
|
|
|
|
|
|
async def rule_detail(user_id: int, rule: Rule, system_ids: list[int] | None = None) -> dict:
|
|
"""The full record, with its areas and edges attached.
|
|
|
|
ONE seam for both doors and every write path, so create, update and get
|
|
cannot disagree about what a rule looks like coming back — the same
|
|
reasoning as attach_relations for notes (#2859), and the same reasoning
|
|
rule_brief exists for one level down.
|
|
|
|
`system_ids=None` means "leave the tags alone"; a list (including [])
|
|
REPLACES them.
|
|
"""
|
|
if system_ids is not None:
|
|
await set_rule_systems(rule.id, user_id, system_ids)
|
|
data = rule.to_dict()
|
|
systems = (await list_rule_systems([rule.id])).get(rule.id, [])
|
|
relations = (await list_rule_relations([rule.id])).get(rule.id, [])
|
|
# Attached only when present (#2483): an empty key reads as a capability
|
|
# the record has and isn't using, which is a different claim.
|
|
if systems:
|
|
data["systems"] = systems
|
|
if relations:
|
|
data["relations"] = relations
|
|
return data
|
|
|
|
|
|
async def create_rule(
|
|
topic_id: int, user_id: int, title: str, statement: str,
|
|
why: str = "", how_to_apply: str = "", order_index: int = 0,
|
|
when_to_apply: str = "", tier: str = "always_on", arose_from_id: int = 0,
|
|
verify_with: str = "", expires_when: str = "",
|
|
) -> Rule:
|
|
async with async_session() as session:
|
|
await _assert_topic_owned(session, topic_id, user_id)
|
|
rule = Rule(
|
|
topic_id=topic_id,
|
|
title=title,
|
|
statement=statement,
|
|
when_to_apply=when_to_apply or None,
|
|
tier=_valid_tier(tier),
|
|
why=why or None,
|
|
how_to_apply=how_to_apply or None,
|
|
verify_with=verify_with or None,
|
|
expires_when=expires_when or None,
|
|
arose_from_id=arose_from_id or None,
|
|
order_index=order_index,
|
|
)
|
|
session.add(rule)
|
|
await session.commit()
|
|
await session.refresh(rule)
|
|
_refresh_rule_embedding(rule)
|
|
return rule
|
|
|
|
|
|
async def create_project_rule(
|
|
project_id: int, user_id: int, title: str, statement: str,
|
|
why: str = "", how_to_apply: str = "", order_index: int = 0,
|
|
when_to_apply: str = "", tier: str = "always_on", arose_from_id: int = 0,
|
|
verify_with: str = "", expires_when: str = "",
|
|
) -> Rule:
|
|
"""Create a rule scoped to a single project (no rulebook ceremony).
|
|
|
|
Project-scoped rules apply only to the named project; they don't
|
|
propagate via rulebook subscriptions. Topic_id is left NULL — the
|
|
CHECK constraint enforces exactly-one of (topic_id, project_id).
|
|
"""
|
|
async with async_session() as session:
|
|
await _assert_project_owned(session, project_id, user_id)
|
|
rule = Rule(
|
|
project_id=project_id,
|
|
title=title,
|
|
statement=statement,
|
|
when_to_apply=when_to_apply or None,
|
|
tier=_valid_tier(tier),
|
|
why=why or None,
|
|
how_to_apply=how_to_apply or None,
|
|
verify_with=verify_with or None,
|
|
expires_when=expires_when or None,
|
|
arose_from_id=arose_from_id or None,
|
|
order_index=order_index,
|
|
)
|
|
session.add(rule)
|
|
await session.commit()
|
|
await session.refresh(rule)
|
|
_refresh_rule_embedding(rule)
|
|
return rule
|
|
|
|
|
|
async def list_rules(
|
|
user_id: int,
|
|
rulebook_id: int | None = None,
|
|
topic_id: int | None = None,
|
|
project_id: int | None = None,
|
|
) -> list[Rule]:
|
|
"""List rules filtered by any of the three IDs. All filters are ownership-scoped.
|
|
|
|
When project_id is set, the result includes both rulebook rules reached via
|
|
project_rulebook_subscriptions AND project-scoped rules (Rule.project_id).
|
|
When rulebook_id or topic_id is set, project-scoped rules are excluded by
|
|
construction (they have neither). With no filter, only rulebook rules are
|
|
returned — adding all of a user's project-scoped rules unprompted would
|
|
surprise existing callers.
|
|
"""
|
|
from scribe.models.rulebook import project_rulebook_subscriptions
|
|
|
|
async with async_session() as session:
|
|
stmt = (
|
|
select(Rule)
|
|
.join(RulebookTopic, Rule.topic_id == RulebookTopic.id)
|
|
.join(Rulebook, RulebookTopic.rulebook_id == Rulebook.id)
|
|
.where(
|
|
Rulebook.owner_user_id == user_id,
|
|
Rule.deleted_at.is_(None),
|
|
RulebookTopic.deleted_at.is_(None),
|
|
Rulebook.deleted_at.is_(None),
|
|
)
|
|
)
|
|
if topic_id:
|
|
stmt = stmt.where(Rule.topic_id == topic_id)
|
|
if rulebook_id:
|
|
stmt = stmt.where(RulebookTopic.rulebook_id == rulebook_id)
|
|
if project_id:
|
|
stmt = (
|
|
stmt.join(
|
|
project_rulebook_subscriptions,
|
|
project_rulebook_subscriptions.c.rulebook_id == Rulebook.id,
|
|
)
|
|
.where(project_rulebook_subscriptions.c.project_id == project_id)
|
|
)
|
|
stmt = stmt.order_by(
|
|
Rulebook.id, RulebookTopic.order_index, Rule.order_index, Rule.title,
|
|
)
|
|
result = await session.execute(stmt)
|
|
rulebook_rules = list(result.scalars().all())
|
|
|
|
if not project_id:
|
|
return rulebook_rules
|
|
|
|
# Project-scoped rules (topic_id IS NULL, project_id matches).
|
|
# Verifies ownership by joining Project on user_id.
|
|
from scribe.models.project import Project
|
|
proj_stmt = (
|
|
select(Rule)
|
|
.join(Project, Rule.project_id == Project.id)
|
|
.where(
|
|
Project.user_id == user_id,
|
|
Rule.project_id == project_id,
|
|
Rule.deleted_at.is_(None),
|
|
Project.deleted_at.is_(None),
|
|
)
|
|
.order_by(Rule.order_index, Rule.title)
|
|
)
|
|
proj_result = await session.execute(proj_stmt)
|
|
return rulebook_rules + list(proj_result.scalars().all())
|
|
|
|
|
|
def _excluded_rulebook_ids_q(project_id: int):
|
|
"""Subquery: the always-on rulebooks this project opted out of at
|
|
inception (milestone 297) — used by every rule-resolution path so an
|
|
exclusion is total, not just cosmetic."""
|
|
from scribe.models.rulebook import project_rulebook_exclusions
|
|
|
|
return select(project_rulebook_exclusions.c.rulebook_id).where(
|
|
project_rulebook_exclusions.c.project_id == project_id
|
|
)
|
|
|
|
|
|
async def excluded_always_on_rulebooks(user_id: int, project_id: int) -> list[dict]:
|
|
"""[{id, title}] of the always-on rulebooks excluded for ``project_id``
|
|
(owner-scoped). Empty for an undecided or inherit-all project."""
|
|
from scribe.models.rulebook import project_rulebook_exclusions
|
|
|
|
if not project_id:
|
|
return []
|
|
async with async_session() as session:
|
|
rows = (
|
|
await session.execute(
|
|
select(Rulebook.id, Rulebook.title)
|
|
.join(project_rulebook_exclusions,
|
|
project_rulebook_exclusions.c.rulebook_id == Rulebook.id)
|
|
.where(
|
|
project_rulebook_exclusions.c.project_id == project_id,
|
|
Rulebook.owner_user_id == user_id,
|
|
Rulebook.deleted_at.is_(None),
|
|
)
|
|
.order_by(Rulebook.title)
|
|
)
|
|
).all()
|
|
return [{"id": rid, "title": title} for rid, title in rows]
|
|
|
|
|
|
async def list_always_on_rules(
|
|
user_id: int, limit: int = 100, project_id: int = 0,
|
|
) -> list[Rule]:
|
|
"""Return all rules from rulebooks flagged always_on for the user.
|
|
|
|
Called by the MCP tool of the same name at session start to load the
|
|
standing rules that apply regardless of which project (if any) is in
|
|
scope. Ordering matches list_rules so results are stable across calls.
|
|
|
|
``project_id`` (milestone 297): inside a project that excluded specific
|
|
always-on rulebooks at inception, those rulebooks' rules are NOT
|
|
returned — the project decided not to inherit them. 0 = the user-wide
|
|
set, which is what a session sees before a project is in scope.
|
|
"""
|
|
async with async_session() as session:
|
|
q = (
|
|
select(Rule)
|
|
.join(RulebookTopic, Rule.topic_id == RulebookTopic.id)
|
|
.join(Rulebook, RulebookTopic.rulebook_id == Rulebook.id)
|
|
.where(
|
|
Rulebook.owner_user_id == user_id,
|
|
Rulebook.always_on.is_(True),
|
|
Rule.deleted_at.is_(None),
|
|
RulebookTopic.deleted_at.is_(None),
|
|
Rulebook.deleted_at.is_(None),
|
|
# TIER (milestone 307). This is the SESSION-START call, made
|
|
# before any project is in scope — there is no area vocabulary
|
|
# to match a conditional rule against yet, so only the
|
|
# unconditional tier belongs here. A conditional rule reaches a
|
|
# session through enter_project (by area) or search (by
|
|
# meaning), not by being resident.
|
|
#
|
|
# Behaviour is unchanged until rules are actually re-tiered:
|
|
# `tier` defaults to always_on, so every existing rule still
|
|
# arrives exactly as it did.
|
|
Rule.tier == "always_on",
|
|
)
|
|
)
|
|
if project_id:
|
|
q = q.where(Rulebook.id.notin_(_excluded_rulebook_ids_q(project_id)))
|
|
result = await session.execute(
|
|
q.order_by(
|
|
Rulebook.id, RulebookTopic.order_index, Rule.order_index, Rule.title,
|
|
).limit(limit)
|
|
)
|
|
return list(result.scalars().all())
|
|
|
|
|
|
async def _fetch_owned_rule(session, rule_id: int, user_id: int) -> Optional[Rule]:
|
|
"""Fetch a rule by id, scoped to user owning either its rulebook
|
|
(via topic) or its project (via project_id). Honors soft-delete.
|
|
Returns None when not found or not owned.
|
|
"""
|
|
from scribe.models.project import Project
|
|
|
|
# Path A — rulebook rule.
|
|
rulebook_rule = (await session.execute(
|
|
select(Rule)
|
|
.join(RulebookTopic, Rule.topic_id == RulebookTopic.id)
|
|
.join(Rulebook, RulebookTopic.rulebook_id == Rulebook.id)
|
|
.where(
|
|
Rule.id == rule_id,
|
|
Rulebook.owner_user_id == user_id,
|
|
Rule.deleted_at.is_(None),
|
|
RulebookTopic.deleted_at.is_(None),
|
|
Rulebook.deleted_at.is_(None),
|
|
)
|
|
)).scalar_one_or_none()
|
|
if rulebook_rule is not None:
|
|
return rulebook_rule
|
|
|
|
# Path B — project-scoped rule.
|
|
project_rule = (await session.execute(
|
|
select(Rule)
|
|
.join(Project, Rule.project_id == Project.id)
|
|
.where(
|
|
Rule.id == rule_id,
|
|
Project.user_id == user_id,
|
|
Rule.deleted_at.is_(None),
|
|
Project.deleted_at.is_(None),
|
|
)
|
|
)).scalar_one_or_none()
|
|
return project_rule
|
|
|
|
|
|
async def get_rule(rule_id: int, user_id: int) -> Optional[Rule]:
|
|
async with async_session() as session:
|
|
return await _fetch_owned_rule(session, rule_id, user_id)
|
|
|
|
|
|
async def update_rule(
|
|
rule_id: int, user_id: int, clear: Iterable[str] = (), **fields,
|
|
) -> Optional[Rule]:
|
|
"""Patch a rule. `clear` names fields to unset; **fields carries new values.
|
|
|
|
Clearing is EXPLICIT and separate because a nullable field cannot be
|
|
emptied by passing it. The MCP door reads "" as "leave this alone" — an
|
|
agent filling three fields must not wipe the other five — so a caller
|
|
there has no value that means "remove it", and a rule that stops being a
|
|
constraint genuinely needs its check removed. Naming the field is the one
|
|
form that cannot happen by accident.
|
|
|
|
Callers that DO have a meaningful empty value (the REST door, where a
|
|
cleared form input arrives as "") get the same outcome through
|
|
NULLABLE_RULE_TEXT normalisation below, so the two doors keep their own
|
|
idiom and agree about the result.
|
|
"""
|
|
async with async_session() as session:
|
|
rule = await _fetch_owned_rule(session, rule_id, user_id)
|
|
if rule is None:
|
|
return None
|
|
allowed = {
|
|
"title", "statement", "why", "how_to_apply", "order_index",
|
|
"when_to_apply", "tier", "arose_from_id",
|
|
"verify_with", "expires_when",
|
|
}
|
|
check_before = rule.verify_with
|
|
for key in clear:
|
|
if key in allowed and key in NULLABLE_RULE_TEXT:
|
|
setattr(rule, key, None)
|
|
elif key == "arose_from_id":
|
|
setattr(rule, key, None)
|
|
for key, value in fields.items():
|
|
if key not in allowed or value is None:
|
|
continue
|
|
if key == "tier":
|
|
value = _valid_tier(value)
|
|
elif key in NULLABLE_RULE_TEXT:
|
|
value = value or None
|
|
elif key == "arose_from_id":
|
|
value = value or None
|
|
setattr(rule, key, value)
|
|
# A verification stamp certifies A CHECK, not a rule. Rewrite or
|
|
# remove the check and the old stamp certifies something that no
|
|
# longer exists — so it is dropped, and the rule re-enters the sweep.
|
|
# The safe direction, for the same reason _valid_tier falls back to
|
|
# always_on: a rule wrongly listed as due costs one look, a rule
|
|
# wrongly vouched for costs the thing the sweep exists to catch.
|
|
if rule.verify_with != check_before:
|
|
rule.verified_at = None
|
|
await session.commit()
|
|
await session.refresh(rule)
|
|
_refresh_rule_embedding(rule)
|
|
return rule
|
|
|
|
|
|
# ── Canon tags + typed edges (milestone 307) ───────────────────────────
|
|
|
|
async def set_rule_systems(
|
|
rule_id: int, user_id: int, canonical_ids: list[int],
|
|
) -> list[int] | None:
|
|
"""Replace which global AREAS a rule is about. None if not owned.
|
|
|
|
Set-semantics like set_record_systems: the list given IS the state after,
|
|
so an empty list clears the tags. Points at the canonical catalog, never a
|
|
project's System — a family rule tagged to one project's row would bind
|
|
itself to that project's vocabulary.
|
|
"""
|
|
from scribe.models.canonical_system import CanonicalSystem
|
|
from scribe.models.rulebook import rule_systems as rule_systems_t
|
|
|
|
async with async_session() as session:
|
|
rule = await _fetch_owned_rule(session, rule_id, user_id)
|
|
if rule is None:
|
|
return None
|
|
wanted = set(canonical_ids or [])
|
|
if wanted:
|
|
live = set((await session.execute(
|
|
select(CanonicalSystem.id).where(
|
|
CanonicalSystem.id.in_(wanted),
|
|
CanonicalSystem.deleted_at.is_(None),
|
|
)
|
|
)).scalars().all())
|
|
# Silently dropping an unknown id would leave the caller believing
|
|
# a tag exists; keep only the live ones and report what stuck.
|
|
wanted &= live
|
|
await session.execute(
|
|
sql_delete(rule_systems_t).where(rule_systems_t.c.rule_id == rule_id)
|
|
)
|
|
for canonical_id in sorted(wanted):
|
|
await session.execute(
|
|
insert(rule_systems_t).values(rule_id=rule_id, canonical_id=canonical_id)
|
|
)
|
|
await session.commit()
|
|
return sorted(wanted)
|
|
|
|
|
|
async def list_rule_systems(rule_ids: list[int]) -> dict[int, list[dict]]:
|
|
"""The canon tags for a batch of rules, keyed by rule id.
|
|
|
|
Batched on purpose: the surfacing paths ask about a whole payload of rules
|
|
at once, and one query per rule would turn every session start into an
|
|
N+1.
|
|
"""
|
|
from scribe.models.canonical_system import CanonicalSystem
|
|
from scribe.models.rulebook import rule_systems as rule_systems_t
|
|
|
|
if not rule_ids:
|
|
return {}
|
|
async with async_session() as session:
|
|
rows = (await session.execute(
|
|
select(rule_systems_t.c.rule_id, CanonicalSystem.id, CanonicalSystem.name)
|
|
.join(CanonicalSystem, CanonicalSystem.id == rule_systems_t.c.canonical_id)
|
|
.where(
|
|
rule_systems_t.c.rule_id.in_(rule_ids),
|
|
CanonicalSystem.deleted_at.is_(None),
|
|
)
|
|
.order_by(CanonicalSystem.order_index)
|
|
)).all()
|
|
out: dict[int, list[dict]] = {}
|
|
for rule_id, canonical_id, name in rows:
|
|
out.setdefault(rule_id, []).append({"id": canonical_id, "name": name})
|
|
return out
|
|
|
|
|
|
async def add_rule_relation(
|
|
user_id: int, from_rule_id: int, to_rule_id: int, kind: str, note: str = "",
|
|
) -> RuleRelation | None:
|
|
"""Draw a typed edge between two rules. None if either isn't owned.
|
|
|
|
Both ends are ownership-checked: an edge is only meaningful if the drawer
|
|
can see both rules, and a one-sided edge would surface a rule the caller
|
|
has no business reading.
|
|
|
|
Idempotent — re-drawing an existing edge returns it rather than raising, so
|
|
a true-up pass can be re-run without cleaning up first.
|
|
"""
|
|
if kind not in RELATION_KINDS:
|
|
raise ValueError(f"kind must be one of {RELATION_KINDS}, got {kind!r}")
|
|
if from_rule_id == to_rule_id:
|
|
raise ValueError("a rule cannot relate to itself")
|
|
async with async_session() as session:
|
|
for rid in (from_rule_id, to_rule_id):
|
|
if await _fetch_owned_rule(session, rid, user_id) is None:
|
|
return None
|
|
existing = await session.scalar(
|
|
select(RuleRelation).where(
|
|
RuleRelation.from_rule_id == from_rule_id,
|
|
RuleRelation.to_rule_id == to_rule_id,
|
|
RuleRelation.kind == kind,
|
|
)
|
|
)
|
|
if existing is not None:
|
|
return existing
|
|
relation = RuleRelation(
|
|
from_rule_id=from_rule_id, to_rule_id=to_rule_id,
|
|
kind=kind, note=note or None,
|
|
)
|
|
session.add(relation)
|
|
await session.commit()
|
|
await session.refresh(relation)
|
|
return relation
|
|
|
|
|
|
async def remove_rule_relation(user_id: int, relation_id: int) -> bool:
|
|
async with async_session() as session:
|
|
relation = await session.get(RuleRelation, relation_id)
|
|
if relation is None:
|
|
return False
|
|
if await _fetch_owned_rule(session, relation.from_rule_id, user_id) is None:
|
|
return False
|
|
await session.delete(relation)
|
|
await session.commit()
|
|
return True
|
|
|
|
|
|
async def list_rule_relations(rule_ids: list[int]) -> dict[int, list[dict]]:
|
|
"""Edges touching a batch of rules, keyed by rule id.
|
|
|
|
`co_surfaces` is reported from BOTH ends off a single stored row — it means
|
|
"these fail together", which is not a claim with a direction. The other two
|
|
are directional and are reported as stored, with `direction` naming which
|
|
end this rule is: an override read from the wrong end would invert what it
|
|
says.
|
|
"""
|
|
if not rule_ids:
|
|
return {}
|
|
async with async_session() as session:
|
|
rows = (await session.execute(
|
|
select(RuleRelation).where(
|
|
(RuleRelation.from_rule_id.in_(rule_ids))
|
|
| (RuleRelation.to_rule_id.in_(rule_ids))
|
|
)
|
|
)).scalars().all()
|
|
out: dict[int, list[dict]] = {}
|
|
wanted = set(rule_ids)
|
|
for relation in rows:
|
|
if relation.from_rule_id in wanted:
|
|
out.setdefault(relation.from_rule_id, []).append({
|
|
"id": relation.id, "kind": relation.kind,
|
|
"rule_id": relation.to_rule_id,
|
|
"direction": "outgoing", "note": relation.note or "",
|
|
})
|
|
if relation.to_rule_id in wanted:
|
|
out.setdefault(relation.to_rule_id, []).append({
|
|
"id": relation.id, "kind": relation.kind,
|
|
"rule_id": relation.from_rule_id,
|
|
"direction": "incoming", "note": relation.note or "",
|
|
})
|
|
return out
|
|
|
|
|
|
async def delete_rule(rule_id: int, user_id: int) -> None:
|
|
async with async_session() as session:
|
|
rule = await _fetch_owned_rule(session, rule_id, user_id)
|
|
if rule is None:
|
|
return
|
|
await session.delete(rule)
|
|
await session.commit()
|
|
|
|
|
|
# ── Subscriptions + get_applicable_rules ───────────────────────────────
|
|
|
|
from sqlalchemy.exc import IntegrityError
|
|
|
|
|
|
async def subscribe_project(
|
|
project_id: int, rulebook_id: int, user_id: int,
|
|
) -> None:
|
|
"""Add a subscription. Idempotent — duplicates raise; we swallow."""
|
|
from scribe.models.rulebook import project_rulebook_subscriptions
|
|
|
|
async with async_session() as session:
|
|
await _assert_project_owned(session, project_id, user_id)
|
|
await _assert_rulebook_owned(session, rulebook_id, user_id)
|
|
# ON CONFLICT DO NOTHING via try/except to keep dialect-agnostic.
|
|
try:
|
|
await session.execute(
|
|
insert(project_rulebook_subscriptions).values(
|
|
project_id=project_id, rulebook_id=rulebook_id,
|
|
)
|
|
)
|
|
await session.commit()
|
|
except Exception:
|
|
await session.rollback() # PK collision = already subscribed; fine.
|
|
|
|
|
|
async def unsubscribe_project(
|
|
project_id: int, rulebook_id: int, user_id: int,
|
|
) -> None:
|
|
from scribe.models.rulebook import project_rulebook_subscriptions
|
|
|
|
async with async_session() as session:
|
|
await _assert_project_owned(session, project_id, user_id)
|
|
await _assert_rulebook_owned(session, rulebook_id, user_id)
|
|
await session.execute(
|
|
sql_delete(project_rulebook_subscriptions).where(
|
|
project_rulebook_subscriptions.c.project_id == project_id,
|
|
project_rulebook_subscriptions.c.rulebook_id == rulebook_id,
|
|
)
|
|
)
|
|
await session.commit()
|
|
|
|
|
|
# ── Suppressions — project-level mute of rulebook rules / topics ────────
|
|
|
|
async def suppress_rule_for_project(
|
|
project_id: int, rule_id: int, user_id: int,
|
|
) -> None:
|
|
"""Mute one rulebook rule for one project. Idempotent."""
|
|
from scribe.models.rulebook import project_rule_suppressions
|
|
|
|
async with async_session() as session:
|
|
await _assert_project_owned(session, project_id, user_id)
|
|
await _assert_rulebook_rule_owned(session, rule_id, user_id)
|
|
try:
|
|
await session.execute(
|
|
insert(project_rule_suppressions).values(
|
|
project_id=project_id, rule_id=rule_id,
|
|
)
|
|
)
|
|
await session.commit()
|
|
except Exception:
|
|
await session.rollback() # PK collision = already suppressed; fine.
|
|
|
|
|
|
async def unsuppress_rule_for_project(
|
|
project_id: int, rule_id: int, user_id: int,
|
|
) -> None:
|
|
"""Unmute one rulebook rule for one project. Idempotent."""
|
|
from scribe.models.rulebook import project_rule_suppressions
|
|
|
|
async with async_session() as session:
|
|
await _assert_project_owned(session, project_id, user_id)
|
|
await session.execute(
|
|
sql_delete(project_rule_suppressions).where(
|
|
project_rule_suppressions.c.project_id == project_id,
|
|
project_rule_suppressions.c.rule_id == rule_id,
|
|
)
|
|
)
|
|
await session.commit()
|
|
|
|
|
|
async def exclude_always_on_rulebook_for_project(
|
|
project_id: int, rulebook_id: int, user_id: int,
|
|
) -> None:
|
|
"""Opt one project out of a whole ALWAYS-ON rulebook (milestone 297).
|
|
Owner-only on both sides; the rulebook must be always_on — a subscribed
|
|
rulebook is left by unsubscribing, not excluding. Idempotent."""
|
|
from scribe.models.rulebook import project_rulebook_exclusions
|
|
|
|
async with async_session() as session:
|
|
await _assert_project_owned(session, project_id, user_id)
|
|
await _assert_rulebook_owned(session, rulebook_id, user_id)
|
|
rb = await session.get(Rulebook, rulebook_id)
|
|
if rb is None or not rb.always_on:
|
|
raise ValueError(
|
|
f"rulebook {rulebook_id} is not always-on — it binds only by "
|
|
"subscription; unsubscribe_project_from_rulebook instead"
|
|
)
|
|
try:
|
|
await session.execute(
|
|
insert(project_rulebook_exclusions).values(
|
|
project_id=project_id, rulebook_id=rulebook_id,
|
|
)
|
|
)
|
|
await session.commit()
|
|
except IntegrityError:
|
|
await session.rollback() # already excluded — idempotent
|
|
|
|
|
|
async def include_always_on_rulebook_for_project(
|
|
project_id: int, rulebook_id: int, user_id: int,
|
|
) -> None:
|
|
"""Undo exclude_always_on_rulebook_for_project. Idempotent."""
|
|
from scribe.models.rulebook import project_rulebook_exclusions
|
|
|
|
async with async_session() as session:
|
|
await _assert_project_owned(session, project_id, user_id)
|
|
await session.execute(
|
|
sql_delete(project_rulebook_exclusions).where(
|
|
project_rulebook_exclusions.c.project_id == project_id,
|
|
project_rulebook_exclusions.c.rulebook_id == rulebook_id,
|
|
)
|
|
)
|
|
await session.commit()
|
|
|
|
|
|
async def suppress_topic_for_project(
|
|
project_id: int, topic_id: int, user_id: int,
|
|
) -> None:
|
|
"""Mute every rule under one topic for one project. Idempotent."""
|
|
from scribe.models.rulebook import project_topic_suppressions
|
|
|
|
async with async_session() as session:
|
|
await _assert_project_owned(session, project_id, user_id)
|
|
await _assert_topic_owned(session, topic_id, user_id)
|
|
try:
|
|
await session.execute(
|
|
insert(project_topic_suppressions).values(
|
|
project_id=project_id, topic_id=topic_id,
|
|
)
|
|
)
|
|
await session.commit()
|
|
except Exception:
|
|
await session.rollback()
|
|
|
|
|
|
async def unsuppress_topic_for_project(
|
|
project_id: int, topic_id: int, user_id: int,
|
|
) -> None:
|
|
"""Unmute a topic for one project. Idempotent."""
|
|
from scribe.models.rulebook import project_topic_suppressions
|
|
|
|
async with async_session() as session:
|
|
await _assert_project_owned(session, project_id, user_id)
|
|
await session.execute(
|
|
sql_delete(project_topic_suppressions).where(
|
|
project_topic_suppressions.c.project_id == project_id,
|
|
project_topic_suppressions.c.topic_id == topic_id,
|
|
)
|
|
)
|
|
await session.commit()
|
|
|
|
|
|
async def get_applicable_rules(
|
|
project_id: int, user_id: int, limit: int = 50,
|
|
) -> dict:
|
|
"""Return rules applicable to a project — both via rulebook subscriptions
|
|
and project-scoped rules (Rule.project_id matches), with suppressed rules
|
|
and suppressed topics filtered out.
|
|
|
|
Shape:
|
|
{
|
|
"rules": [{id, title, statement,
|
|
topic_id, topic_title,
|
|
rulebook_id, rulebook_title}, ...],
|
|
"project_rules": [{id, title, statement}, ...],
|
|
"suppressed_rules": [{id, title,
|
|
topic_id, topic_title,
|
|
rulebook_id, rulebook_title}, ...],
|
|
"suppressed_topics": [{id, title,
|
|
rulebook_id, rulebook_title}, ...],
|
|
"truncated": bool,
|
|
"subscribed_rulebooks": [{id, title}, ...]
|
|
}
|
|
|
|
`rules` is the subscription-derived set with project-level suppressions
|
|
applied. `project_rules` is the project-scoped set (never suppressed —
|
|
delete instead). `suppressed_rules` / `suppressed_topics` carry the
|
|
titles + rulebook context callers need to display what was filtered
|
|
without round-tripping for names.
|
|
"""
|
|
from scribe.models.rulebook import (
|
|
project_rulebook_subscriptions,
|
|
project_rule_suppressions,
|
|
project_topic_suppressions,
|
|
)
|
|
|
|
async with async_session() as session:
|
|
# Subscribed rulebooks for the project (ownership-scoped).
|
|
sub_q = (
|
|
select(Rulebook.id, Rulebook.title)
|
|
.join(
|
|
project_rulebook_subscriptions,
|
|
project_rulebook_subscriptions.c.rulebook_id == Rulebook.id,
|
|
)
|
|
.where(
|
|
project_rulebook_subscriptions.c.project_id == project_id,
|
|
Rulebook.owner_user_id == user_id,
|
|
Rulebook.deleted_at.is_(None),
|
|
)
|
|
.order_by(Rulebook.title)
|
|
)
|
|
sub_rows = (await session.execute(sub_q)).all()
|
|
subscribed_rulebooks = [
|
|
{"id": rb_id, "title": rb_title} for rb_id, rb_title in sub_rows
|
|
]
|
|
|
|
# Suppressed rules — joined to topic + rulebook so callers can render
|
|
# context without a follow-up lookup. Ownership-scoped via rulebook.
|
|
suppressed_rules_q = (
|
|
select(
|
|
Rule.id, Rule.title,
|
|
RulebookTopic.id.label("topic_id"),
|
|
RulebookTopic.title.label("topic_title"),
|
|
Rulebook.id.label("rulebook_id"),
|
|
Rulebook.title.label("rulebook_title"),
|
|
)
|
|
.join(project_rule_suppressions, project_rule_suppressions.c.rule_id == Rule.id)
|
|
.join(RulebookTopic, Rule.topic_id == RulebookTopic.id)
|
|
.join(Rulebook, RulebookTopic.rulebook_id == Rulebook.id)
|
|
.where(
|
|
project_rule_suppressions.c.project_id == project_id,
|
|
Rulebook.owner_user_id == user_id,
|
|
)
|
|
.order_by(Rulebook.title, RulebookTopic.title, Rule.title)
|
|
)
|
|
suppressed_rule_rows = (await session.execute(suppressed_rules_q)).all()
|
|
suppressed_rules = [
|
|
{"id": rid, "title": rt, "topic_id": ti, "topic_title": tt,
|
|
"rulebook_id": rbi, "rulebook_title": rbt}
|
|
for rid, rt, ti, tt, rbi, rbt in suppressed_rule_rows
|
|
]
|
|
suppressed_rule_ids = [r["id"] for r in suppressed_rules]
|
|
|
|
# Suppressed topics — joined to rulebook for context.
|
|
suppressed_topics_q = (
|
|
select(
|
|
RulebookTopic.id, RulebookTopic.title,
|
|
Rulebook.id.label("rulebook_id"),
|
|
Rulebook.title.label("rulebook_title"),
|
|
)
|
|
.join(project_topic_suppressions, project_topic_suppressions.c.topic_id == RulebookTopic.id)
|
|
.join(Rulebook, RulebookTopic.rulebook_id == Rulebook.id)
|
|
.where(
|
|
project_topic_suppressions.c.project_id == project_id,
|
|
Rulebook.owner_user_id == user_id,
|
|
)
|
|
.order_by(Rulebook.title, RulebookTopic.title)
|
|
)
|
|
suppressed_topic_rows = (await session.execute(suppressed_topics_q)).all()
|
|
suppressed_topics = [
|
|
{"id": tid, "title": tt, "rulebook_id": rbi, "rulebook_title": rbt}
|
|
for tid, tt, rbi, rbt in suppressed_topic_rows
|
|
]
|
|
suppressed_topic_ids = [t["id"] for t in suppressed_topics]
|
|
|
|
# Applicable rules (limit + 1 so we can detect truncation). Filter
|
|
# in SQL so truncation reflects the post-suppression count, not the
|
|
# raw subscription count.
|
|
# Selects the ENTITY, not a column list: rule_brief is the one place
|
|
# that decides which fields a surfaced rule carries, and a column list
|
|
# here would be a second such decision to keep in step. The row count
|
|
# is bounded by `limit`, so this is a listing, not a scan.
|
|
rules_q = (
|
|
select(
|
|
Rule,
|
|
RulebookTopic.title.label("topic_title"),
|
|
Rulebook.id.label("rulebook_id"),
|
|
Rulebook.title.label("rulebook_title"),
|
|
)
|
|
.join(RulebookTopic, Rule.topic_id == RulebookTopic.id)
|
|
.join(Rulebook, RulebookTopic.rulebook_id == Rulebook.id)
|
|
.join(
|
|
project_rulebook_subscriptions,
|
|
project_rulebook_subscriptions.c.rulebook_id == Rulebook.id,
|
|
)
|
|
.where(
|
|
project_rulebook_subscriptions.c.project_id == project_id,
|
|
Rulebook.owner_user_id == user_id,
|
|
Rule.deleted_at.is_(None),
|
|
RulebookTopic.deleted_at.is_(None),
|
|
Rulebook.deleted_at.is_(None),
|
|
# An inception exclusion is total (milestone 297): a rulebook the
|
|
# project opted out of contributes nothing, subscribed or not.
|
|
Rulebook.id.notin_(_excluded_rulebook_ids_q(project_id)),
|
|
)
|
|
.order_by(
|
|
Rulebook.id, RulebookTopic.order_index, Rule.order_index, Rule.title,
|
|
)
|
|
.limit(limit + 1)
|
|
)
|
|
if suppressed_rule_ids:
|
|
rules_q = rules_q.where(Rule.id.notin_(suppressed_rule_ids))
|
|
if suppressed_topic_ids:
|
|
rules_q = rules_q.where(Rule.topic_id.notin_(suppressed_topic_ids))
|
|
# TIER (milestone 307). always_on rules are resident, as every rule was
|
|
# before tiers existed. A conditional rule is REACHABLE, and reaches
|
|
# this project only when it is tagged to an area this project actually
|
|
# works in — a deterministic tag match, never a similarity score, so
|
|
# bindingness never depends on a ranking (D7).
|
|
#
|
|
# Applied in SQL rather than by filtering afterwards, so `limit` counts
|
|
# the rules that will actually be surfaced instead of counting rules
|
|
# that are about to be dropped.
|
|
project_area_ids = (await session.execute(
|
|
select(System.canonical_id).where(
|
|
System.project_id == project_id,
|
|
System.canonical_id.is_not(None),
|
|
System.deleted_at.is_(None),
|
|
System.status == "active",
|
|
).distinct()
|
|
)).scalars().all()
|
|
reachable = select(rule_systems.c.rule_id).where(
|
|
rule_systems.c.canonical_id.in_(project_area_ids)
|
|
) if project_area_ids else None
|
|
tier_clause = (Rule.tier == "always_on")
|
|
if reachable is not None:
|
|
tier_clause = or_(tier_clause, Rule.id.in_(reachable))
|
|
rules_q = rules_q.where(tier_clause)
|
|
rule_rows = (await session.execute(rules_q)).all()
|
|
truncated = len(rule_rows) > limit
|
|
rules = [
|
|
rule_brief(rule, topic_title=tt, rulebook_id=rbi, rulebook_title=rbt)
|
|
for rule, tt, rbi, rbt in rule_rows[:limit]
|
|
]
|
|
|
|
# Project-scoped rules — verifies ownership via Project.user_id.
|
|
from scribe.models.project import Project
|
|
proj_rules_q = (
|
|
select(Rule)
|
|
.join(Project, Rule.project_id == Project.id)
|
|
.where(
|
|
Project.user_id == user_id,
|
|
Rule.project_id == project_id,
|
|
Rule.deleted_at.is_(None),
|
|
Project.deleted_at.is_(None),
|
|
)
|
|
.order_by(Rule.order_index, Rule.title)
|
|
)
|
|
if reachable is not None:
|
|
proj_rules_q = proj_rules_q.where(
|
|
or_(Rule.tier == "always_on", Rule.id.in_(reachable))
|
|
)
|
|
else:
|
|
proj_rules_q = proj_rules_q.where(Rule.tier == "always_on")
|
|
proj_rule_rows = (await session.execute(proj_rules_q)).all()
|
|
project_rules = [rule_brief(rule) for (rule,) in proj_rule_rows]
|
|
|
|
# Edges travel with the rules they belong to (milestone 307).
|
|
#
|
|
# A co_surfaces partner that was not otherwise selected is ADDED, because a
|
|
# rule that arrives without the half it fails with is the failure the edge
|
|
# was created to prevent. Suppressions are passed as exclusions so an
|
|
# explicit mute still wins over an edge.
|
|
surfaced_ids = [r["id"] for r in rules] + [r["id"] for r in project_rules]
|
|
partners = await co_surfaced_partners(
|
|
user_id, surfaced_ids, exclude_ids=set(suppressed_rule_ids),
|
|
)
|
|
for partner in partners:
|
|
rules.append(rule_brief(partner, via="co_surfaces"))
|
|
surfaced_ids.append(partner.id)
|
|
|
|
# Relations on every surfaced rule, so a reader can see that an override
|
|
# exists rather than discovering the contradiction by acting on the wrong
|
|
# one. Areas too — they are why a conditional rule is here at all.
|
|
edges = await list_rule_relations(surfaced_ids)
|
|
areas = await list_rule_systems(surfaced_ids)
|
|
for brief in (*rules, *project_rules):
|
|
if edges.get(brief["id"]):
|
|
brief["relations"] = edges[brief["id"]]
|
|
if areas.get(brief["id"]):
|
|
brief["systems"] = areas[brief["id"]]
|
|
|
|
return {
|
|
"rules": rules,
|
|
"project_rules": project_rules,
|
|
"suppressed_rules": suppressed_rules,
|
|
"suppressed_topics": suppressed_topics,
|
|
"truncated": truncated,
|
|
"subscribed_rulebooks": subscribed_rulebooks,
|
|
"excluded_always_on": await excluded_always_on_rulebooks(user_id, project_id),
|
|
}
|
|
|
|
|
|
def rules_payload(applicable: dict) -> dict:
|
|
"""The caller-facing shape of a get_applicable_rules() result.
|
|
|
|
Every surface that hands rules to an agent (enter_project, get_project,
|
|
get_milestone, get_task for legacy plans, start_planning) carries the
|
|
same seven keys under the same names — so a reader learns them once. One
|
|
place renames `rules` → `applicable_rules` and `truncated` →
|
|
`applicable_rules_truncated`; the tools merge this into their payloads.
|
|
`excluded_always_on` (milestone 297) names the always-on rulebooks this
|
|
project decided NOT to inherit, so the departure is visible wherever the
|
|
rules are.
|
|
"""
|
|
return {
|
|
"applicable_rules": applicable["rules"],
|
|
"applicable_rules_truncated": applicable["truncated"],
|
|
"subscribed_rulebooks": applicable["subscribed_rulebooks"],
|
|
"project_rules": applicable.get("project_rules", []),
|
|
"suppressed_rules": applicable.get("suppressed_rules", []),
|
|
"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
|