bvandeusenandClaude Opus 5
91b34619f9
verify_with / expires_when now reach a rule through both doors and come back on every read. The open question this step existed to settle was how to UNSET a nullable field, and the answer is one convention per door: - MCP: "" still means "leave unchanged" — an agent filling three fields must not wipe the other five — so clearing is explicit, clear_fields=["..."]. Naming the field is the one form that cannot happen by accident. - REST: a cleared form input arrives as "", and the service normalises "" to NULL for every nullable rule column, so an emptied input does what it looks like it does. Two idioms, one outcome, and the normalisation is what makes the step-3 sweep correct: `verify_with IS NOT NULL` would otherwise be true for every rule ever touched through the UI, and the sweep would list the whole rulebook and mean nothing. to_dict renders "" and NULL identically, so this is only visible against a real column — hence the integration module rather than a mock. Editing verify_with drops verified_at. A stamp certifies A CHECK, not a rule; reword the check and the old stamp vouches for something that no longer exists. Safe direction, same asymmetry as _valid_tier: a rule wrongly listed as due costs one look, a rule wrongly vouched for costs the thing the sweep exists to catch. Editing anything else leaves the stamp alone, or a rulebook tidy-up would reset every constraint and the ordering would carry nothing. Reads: rule_brief attaches `last_verified` ONLY to a rule that carries a check — its presence is the signal, and it says both "this asserts a fact that can go false" and "here is how long ago anyone confirmed it". "never" rather than null, per #2483. The check text itself stays in get_rule; a listing needs to know which rules can rot, not how to test them. Search hits carry the full trio, since a hit is exactly the moment someone is about to act on a rule. Also folds in the #3078 finding, which had been sitting as a note: create_rule now teaches that when_to_apply is the retrieval surface and must carry the SYMPTOM — the words you would type while stuck — not just the situation. fake_rule gains the three fields as None for the reason the helper already documents one line up: unnamed, verify_with is a truthy MagicMock and every stand-in rule would claim a check it does not have. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
393 lines
14 KiB
Python
393 lines
14 KiB
Python
"""Rulebook / topic REST endpoints.
|
|
|
|
Wraps services/rulebooks.py. Standard Scribe auth: get_current_user_id() is the
|
|
authenticated owner; the service enforces ownership scoping.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from quart import Blueprint, jsonify, request
|
|
|
|
from scribe.auth import get_current_user_id, login_required
|
|
import scribe.services.rulebooks as rulebooks_svc
|
|
from scribe.services.trash import delete as trash_delete
|
|
|
|
rulebooks_bp = Blueprint("rulebooks", __name__, url_prefix="/api")
|
|
|
|
|
|
|
|
# ── Rulebooks ───────────────────────────────────────────────────────────
|
|
|
|
@rulebooks_bp.get("/rulebooks")
|
|
@login_required
|
|
async def list_rulebooks():
|
|
rows = await rulebooks_svc.list_rulebooks(get_current_user_id())
|
|
return jsonify({"rulebooks": [rb.to_dict() for rb in rows]})
|
|
|
|
|
|
@rulebooks_bp.post("/rulebooks")
|
|
@login_required
|
|
async def create_rulebook():
|
|
data = await request.get_json() or {}
|
|
title = (data.get("title") or "").strip()
|
|
if not title:
|
|
return jsonify({"error": "title is required"}), 400
|
|
rb = await rulebooks_svc.create_rulebook(
|
|
user_id=get_current_user_id(),
|
|
title=title,
|
|
description=data.get("description", ""),
|
|
)
|
|
return jsonify(rb.to_dict()), 201
|
|
|
|
|
|
@rulebooks_bp.get("/rulebooks/<int:rulebook_id>")
|
|
@login_required
|
|
async def get_rulebook(rulebook_id: int):
|
|
rb = await rulebooks_svc.get_rulebook(rulebook_id, get_current_user_id())
|
|
if rb is None:
|
|
return jsonify({"error": "rulebook not found"}), 404
|
|
return jsonify(rb.to_dict())
|
|
|
|
|
|
@rulebooks_bp.patch("/rulebooks/<int:rulebook_id>")
|
|
@login_required
|
|
async def update_rulebook(rulebook_id: int):
|
|
data = await request.get_json() or {}
|
|
fields = {k: v for k, v in data.items() if k in ("title", "description", "always_on")}
|
|
rb = await rulebooks_svc.update_rulebook(rulebook_id, get_current_user_id(), **fields)
|
|
if rb is None:
|
|
return jsonify({"error": "rulebook not found"}), 404
|
|
return jsonify(rb.to_dict())
|
|
|
|
|
|
@rulebooks_bp.delete("/rulebooks/<int:rulebook_id>")
|
|
@login_required
|
|
async def delete_rulebook(rulebook_id: int):
|
|
await trash_delete(get_current_user_id(), "rulebook", rulebook_id)
|
|
return "", 204
|
|
|
|
|
|
# ── Topics ──────────────────────────────────────────────────────────────
|
|
|
|
@rulebooks_bp.get("/rulebooks/<int:rulebook_id>/topics")
|
|
@login_required
|
|
async def list_topics(rulebook_id: int):
|
|
try:
|
|
rows = await rulebooks_svc.list_topics(rulebook_id, get_current_user_id())
|
|
except ValueError as exc:
|
|
return jsonify({"error": str(exc)}), 404
|
|
return jsonify({"topics": [t.to_dict() for t in rows]})
|
|
|
|
|
|
@rulebooks_bp.post("/rulebooks/<int:rulebook_id>/topics")
|
|
@login_required
|
|
async def create_topic(rulebook_id: int):
|
|
data = await request.get_json() or {}
|
|
title = (data.get("title") or "").strip()
|
|
if not title:
|
|
return jsonify({"error": "title is required"}), 400
|
|
try:
|
|
topic = await rulebooks_svc.create_topic(
|
|
rulebook_id=rulebook_id,
|
|
user_id=get_current_user_id(),
|
|
title=title,
|
|
description=data.get("description", ""),
|
|
order_index=data.get("order_index", 0),
|
|
)
|
|
except ValueError as exc:
|
|
return jsonify({"error": str(exc)}), 404
|
|
return jsonify(topic.to_dict()), 201
|
|
|
|
|
|
@rulebooks_bp.patch("/rulebook-topics/<int:topic_id>")
|
|
@login_required
|
|
async def update_topic(topic_id: int):
|
|
data = await request.get_json() or {}
|
|
fields = {
|
|
k: v for k, v in data.items()
|
|
if k in ("title", "description", "order_index")
|
|
}
|
|
topic = await rulebooks_svc.update_topic(topic_id, get_current_user_id(), **fields)
|
|
if topic is None:
|
|
return jsonify({"error": "topic not found"}), 404
|
|
return jsonify(topic.to_dict())
|
|
|
|
|
|
@rulebooks_bp.delete("/rulebook-topics/<int:topic_id>")
|
|
@login_required
|
|
async def delete_topic(topic_id: int):
|
|
if await trash_delete(get_current_user_id(), "topic", topic_id) is None:
|
|
return jsonify({"error": "topic not found"}), 404
|
|
return "", 204
|
|
|
|
|
|
# ── Rules ───────────────────────────────────────────────────────────────
|
|
|
|
@rulebooks_bp.get("/rules")
|
|
@login_required
|
|
async def list_rules():
|
|
def _opt_int(name):
|
|
raw = request.args.get(name)
|
|
return int(raw) if raw else None
|
|
|
|
try:
|
|
rulebook_id = _opt_int("rulebook_id")
|
|
topic_id = _opt_int("topic_id")
|
|
project_id = _opt_int("project_id")
|
|
except ValueError:
|
|
return jsonify({"error": "rulebook_id, topic_id, project_id must be integers"}), 400
|
|
|
|
rows = await rulebooks_svc.list_rules(
|
|
user_id=get_current_user_id(),
|
|
rulebook_id=rulebook_id,
|
|
topic_id=topic_id,
|
|
project_id=project_id,
|
|
)
|
|
return jsonify({"rules": [r.to_dict() for r in rows]})
|
|
|
|
|
|
@rulebooks_bp.post("/rulebook-topics/<int:topic_id>/rules")
|
|
@login_required
|
|
async def create_rule(topic_id: int):
|
|
data = await request.get_json() or {}
|
|
title = (data.get("title") or "").strip()
|
|
statement = (data.get("statement") or "").strip()
|
|
if not title or not statement:
|
|
return jsonify({"error": "title and statement are required"}), 400
|
|
try:
|
|
rule = await rulebooks_svc.create_rule(
|
|
topic_id=topic_id,
|
|
user_id=get_current_user_id(),
|
|
title=title,
|
|
statement=statement,
|
|
why=data.get("why", ""),
|
|
how_to_apply=data.get("how_to_apply", ""),
|
|
order_index=data.get("order_index", 0),
|
|
when_to_apply=data.get("when_to_apply", ""),
|
|
tier=data.get("tier", "always_on"),
|
|
arose_from_id=data.get("arose_from_id", 0) or 0,
|
|
verify_with=data.get("verify_with", ""),
|
|
expires_when=data.get("expires_when", ""),
|
|
)
|
|
except ValueError as exc:
|
|
return jsonify({"error": str(exc)}), 404
|
|
return jsonify(await rulebooks_svc.rule_detail(
|
|
get_current_user_id(), rule, data.get("system_ids"),
|
|
)), 201
|
|
|
|
|
|
@rulebooks_bp.get("/rules/<int:rule_id>")
|
|
@login_required
|
|
async def get_rule(rule_id: int):
|
|
uid = get_current_user_id()
|
|
rule = await rulebooks_svc.get_rule(rule_id, uid)
|
|
if rule is None:
|
|
return jsonify({"error": "rule not found"}), 404
|
|
return jsonify(await rulebooks_svc.rule_detail(uid, rule))
|
|
|
|
|
|
@rulebooks_bp.patch("/rules/<int:rule_id>")
|
|
@login_required
|
|
async def update_rule(rule_id: int):
|
|
data = await request.get_json() or {}
|
|
uid = get_current_user_id()
|
|
fields = {
|
|
k: v for k, v in data.items()
|
|
if k in ("title", "statement", "why", "how_to_apply", "order_index",
|
|
"when_to_apply", "tier", "arose_from_id",
|
|
"verify_with", "expires_when")
|
|
}
|
|
# No clear_fields here: a form sends "" for an emptied input, and the
|
|
# service normalises "" to NULL for every nullable text column. The MCP
|
|
# door needs the explicit list only because "" already means "unchanged"
|
|
# there — two idioms, one outcome.
|
|
rule = await rulebooks_svc.update_rule(rule_id, uid, **fields)
|
|
if rule is None:
|
|
return jsonify({"error": "rule not found"}), 404
|
|
return jsonify(await rulebooks_svc.rule_detail(uid, rule, data.get("system_ids")))
|
|
|
|
|
|
@rulebooks_bp.post("/rules/<int:rule_id>/relations")
|
|
@login_required
|
|
async def relate_rules(rule_id: int):
|
|
"""Draw a typed edge FROM this rule to another.
|
|
|
|
Body: {"to_rule_id": N, "kind": "co_surfaces"|"overrides"|"elaborates",
|
|
"note": "..."}. Idempotent — re-drawing an edge returns the existing one.
|
|
"""
|
|
data = await request.get_json() or {}
|
|
to_rule_id = data.get("to_rule_id")
|
|
if not isinstance(to_rule_id, int):
|
|
return jsonify({"error": "to_rule_id is required"}), 400
|
|
try:
|
|
relation = await rulebooks_svc.add_rule_relation(
|
|
get_current_user_id(), rule_id, to_rule_id,
|
|
data.get("kind", ""), data.get("note", ""),
|
|
)
|
|
except ValueError as exc:
|
|
return jsonify({"error": str(exc)}), 400
|
|
if relation is None:
|
|
return jsonify({"error": "rule not found"}), 404
|
|
return jsonify(relation.to_dict()), 201
|
|
|
|
|
|
@rulebooks_bp.delete("/rule-relations/<int:relation_id>")
|
|
@login_required
|
|
async def unrelate_rules(relation_id: int):
|
|
if not await rulebooks_svc.remove_rule_relation(get_current_user_id(), relation_id):
|
|
return jsonify({"error": "relation not found"}), 404
|
|
return "", 204
|
|
|
|
|
|
@rulebooks_bp.delete("/rules/<int:rule_id>")
|
|
@login_required
|
|
async def delete_rule(rule_id: int):
|
|
if await trash_delete(get_current_user_id(), "rule", rule_id) is None:
|
|
return jsonify({"error": "rule not found"}), 404
|
|
return "", 204
|
|
|
|
|
|
# ── Subscriptions ──────────────────────────────────────────────────────
|
|
|
|
@rulebooks_bp.post("/projects/<int:project_id>/rulebook-subscriptions")
|
|
@login_required
|
|
async def subscribe_project(project_id: int):
|
|
data = await request.get_json() or {}
|
|
rulebook_id = data.get("rulebook_id")
|
|
if not rulebook_id:
|
|
return jsonify({"error": "rulebook_id is required"}), 400
|
|
try:
|
|
await rulebooks_svc.subscribe_project(
|
|
project_id=project_id, rulebook_id=int(rulebook_id), user_id=get_current_user_id(),
|
|
)
|
|
except ValueError as exc:
|
|
return jsonify({"error": str(exc)}), 404
|
|
return "", 204
|
|
|
|
|
|
@rulebooks_bp.delete(
|
|
"/projects/<int:project_id>/rulebook-subscriptions/<int:rulebook_id>"
|
|
)
|
|
@login_required
|
|
async def unsubscribe_project(project_id: int, rulebook_id: int):
|
|
try:
|
|
await rulebooks_svc.unsubscribe_project(
|
|
project_id=project_id, rulebook_id=rulebook_id, user_id=get_current_user_id(),
|
|
)
|
|
except ValueError as exc:
|
|
return jsonify({"error": str(exc)}), 404
|
|
return "", 204
|
|
|
|
|
|
@rulebooks_bp.get("/projects/<int:project_id>/rules")
|
|
@login_required
|
|
async def get_project_rules(project_id: int):
|
|
result = await rulebooks_svc.get_applicable_rules(
|
|
project_id=project_id, user_id=get_current_user_id(),
|
|
)
|
|
return jsonify(result)
|
|
|
|
|
|
@rulebooks_bp.post("/projects/<int:project_id>/suppressions/rules/<int:rule_id>")
|
|
@login_required
|
|
async def suppress_project_rule(project_id: int, rule_id: int):
|
|
try:
|
|
await rulebooks_svc.suppress_rule_for_project(
|
|
project_id=project_id, rule_id=rule_id, user_id=get_current_user_id(),
|
|
)
|
|
except ValueError as exc:
|
|
return jsonify({"error": str(exc)}), 404
|
|
return "", 204
|
|
|
|
|
|
@rulebooks_bp.delete("/projects/<int:project_id>/suppressions/rules/<int:rule_id>")
|
|
@login_required
|
|
async def unsuppress_project_rule(project_id: int, rule_id: int):
|
|
try:
|
|
await rulebooks_svc.unsuppress_rule_for_project(
|
|
project_id=project_id, rule_id=rule_id, user_id=get_current_user_id(),
|
|
)
|
|
except ValueError as exc:
|
|
return jsonify({"error": str(exc)}), 404
|
|
return "", 204
|
|
|
|
|
|
@rulebooks_bp.post("/projects/<int:project_id>/suppressions/topics/<int:topic_id>")
|
|
@login_required
|
|
async def suppress_project_topic(project_id: int, topic_id: int):
|
|
try:
|
|
await rulebooks_svc.suppress_topic_for_project(
|
|
project_id=project_id, topic_id=topic_id, user_id=get_current_user_id(),
|
|
)
|
|
except ValueError as exc:
|
|
return jsonify({"error": str(exc)}), 404
|
|
return "", 204
|
|
|
|
|
|
@rulebooks_bp.delete("/projects/<int:project_id>/suppressions/topics/<int:topic_id>")
|
|
@login_required
|
|
async def unsuppress_project_topic(project_id: int, topic_id: int):
|
|
try:
|
|
await rulebooks_svc.unsuppress_topic_for_project(
|
|
project_id=project_id, topic_id=topic_id, user_id=get_current_user_id(),
|
|
)
|
|
except ValueError as exc:
|
|
return jsonify({"error": str(exc)}), 404
|
|
return "", 204
|
|
|
|
|
|
@rulebooks_bp.post("/projects/<int:project_id>/exclusions/rulebooks/<int:rulebook_id>")
|
|
@login_required
|
|
async def exclude_project_rulebook(project_id: int, rulebook_id: int):
|
|
"""Opt the project out of a whole always-on rulebook (milestone 297)."""
|
|
try:
|
|
await rulebooks_svc.exclude_always_on_rulebook_for_project(
|
|
project_id=project_id, rulebook_id=rulebook_id, user_id=get_current_user_id(),
|
|
)
|
|
except ValueError as exc:
|
|
msg = str(exc)
|
|
return jsonify({"error": msg}), (400 if "not always-on" in msg else 404)
|
|
return "", 204
|
|
|
|
|
|
@rulebooks_bp.delete("/projects/<int:project_id>/exclusions/rulebooks/<int:rulebook_id>")
|
|
@login_required
|
|
async def include_project_rulebook(project_id: int, rulebook_id: int):
|
|
try:
|
|
await rulebooks_svc.include_always_on_rulebook_for_project(
|
|
project_id=project_id, rulebook_id=rulebook_id, user_id=get_current_user_id(),
|
|
)
|
|
except ValueError as exc:
|
|
return jsonify({"error": str(exc)}), 404
|
|
return "", 204
|
|
|
|
|
|
@rulebooks_bp.post("/projects/<int:project_id>/rules")
|
|
@login_required
|
|
async def create_project_rule(project_id: int):
|
|
"""Create a rule scoped to a single project. Frontend fast path."""
|
|
data = await request.get_json() or {}
|
|
statement = (data.get("statement") or "").strip()
|
|
if not statement:
|
|
return jsonify({"error": "statement is required"}), 400
|
|
title = (data.get("title") or "").strip() or statement.split(".")[0][:50]
|
|
try:
|
|
rule = await rulebooks_svc.create_project_rule(
|
|
project_id=project_id,
|
|
user_id=get_current_user_id(),
|
|
title=title,
|
|
statement=statement,
|
|
why=data.get("why", ""),
|
|
how_to_apply=data.get("how_to_apply", ""),
|
|
order_index=data.get("order_index", 0),
|
|
when_to_apply=data.get("when_to_apply", ""),
|
|
tier=data.get("tier", "always_on"),
|
|
arose_from_id=data.get("arose_from_id", 0) or 0,
|
|
verify_with=data.get("verify_with", ""),
|
|
expires_when=data.get("expires_when", ""),
|
|
)
|
|
except ValueError as exc:
|
|
return jsonify({"error": str(exc)}), 404
|
|
return jsonify(await rulebooks_svc.rule_detail(
|
|
get_current_user_id(), rule, data.get("system_ids"),
|
|
)), 201
|