feat(rulings): a command or edit touching an area's files shows its rulings, once per session (milestone 444 step 4, #4757)
CI & Build / Python lint (push) Successful in 3s
CI & Build / Plugin hooks (push) Successful in 12s
CI & Build / TypeScript typecheck (push) Successful in 54s
CI & Build / integration (push) Successful in 1m6s
CI & Build / Python tests (push) Failing after 1m22s
CI & Build / Build & push image (push) Skipped

A System's rulings (the Rulings section of its description) now reach the
work by path, not by similarity. Both PreToolUse arms resolve the files a
command or edit names to the Systems whose path_patterns cover them, and the
first touch in a session shows each area's rulings in one line; a repeat is
a one-line reference. A lookup, so no floor, no budget, no retrieval_logs row.

- services/system_rulings: parse_rulings, command_paths (reads and writes,
  relative to the repo root from any cwd; flags, URLs, globs skipped),
  rulings_for_paths
- /tool-rules takes root, cwd and seen_ruling_systems; /prior-art takes
  seen_ruling_systems; both return ruling_system_ids
- hooks share <sid>.rulings.ids (cleared on compaction by the ledger naming
  convention); the Bash hook sends the repo root and cwd
- system_usage_events (migration 0114): surfacings by source, pulls from
  get_system; carried by backup (v20) through the system map
- writing-records: rulings also arrive when the area's files are touched

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-02 23:08:32 -04:00
co-authored by Claude Opus 5.5
parent a113c72b4f
commit 556872c039
17 changed files with 808 additions and 13 deletions
@@ -0,0 +1,45 @@
"""system_usage_events — were an area's rulings read once its files were
touched? (milestone 444 step 4, #4757)
Revision ID: 0114
Revises: 0113
Create Date: 2026-10-03
The usage twin for Systems, beside note_usage_events and rule_usage_events and
separate from both for the same reason they are separate from each other: a
System id restores through its own map. FK-free like its siblings; no CHECK on
`event`, like its siblings.
"""
import sqlalchemy as sa
from alembic import op
revision = "0114"
down_revision = "0113"
branch_labels = None
depends_on = None
def upgrade() -> None:
op.create_table(
"system_usage_events",
sa.Column("id", sa.BigInteger(), primary_key=True),
sa.Column(
"created_at", sa.DateTime(timezone=True), nullable=False,
server_default=sa.text("now()"),
),
sa.Column("user_id", sa.BigInteger(), nullable=True),
sa.Column("system_id", sa.BigInteger(), nullable=False),
sa.Column("event", sa.Text(), nullable=False),
sa.Column("source", sa.Text(), nullable=False),
sa.Column("project_id", sa.BigInteger(), nullable=True),
)
op.create_index(
"ix_system_usage_system_event", "system_usage_events", ["system_id", "event"]
)
op.create_index("ix_system_usage_created_at", "system_usage_events", ["created_at"])
def downgrade() -> None:
op.drop_index("ix_system_usage_created_at", table_name="system_usage_events")
op.drop_index("ix_system_usage_system_event", table_name="system_usage_events")
op.drop_table("system_usage_events")
+1 -1
View File
@@ -1,7 +1,7 @@
{
"name": "scribe",
"description": "Scribe for Claude Code: connects the scribe MCP server, adds the hooks that deliver live project state and relevant records at the right moment, ships the shared client-neutral Scribe skills (using-scribe, writing-plans, reporting-back, systematic-debugging, verification, brainstorming, reusing-code, shape-accounting), and syncs your saved Scribe Processes as skills (/scribe:sync).",
"version": "2026.10.03.0255",
"version": "2026.10.03.0308",
"author": {
"name": "Bryan Van Deusen"
},
+14
View File
@@ -1123,6 +1123,20 @@ scribe_held_query() {
return 0
}
# The Systems whose rulings this session was already shown in full (milestone
# 444), as the `seen_ruling_systems` query both PreToolUse hooks send. A flat
# read, not aged like the rule file: a ruling binds for the whole session, and
# the server already answers a repeat with one short reference line. The file
# follows the ledger name convention, so a compaction clears it and the
# rulings are shown in full again to the context that lost them.
scribe_rulings_query() {
local ids
[ -f "$1" ] || return 0
ids=$(grep -E '^[0-9]+$' "$1" 2>/dev/null | sort -un | tr '\n' ',' | sed 's/,$//')
[ -n "$ids" ] && printf '&seen_ruling_systems=%s' "$ids"
return 0
}
# Drop EVERY per-session ledger, matched by convention rather than listed (#4101).
#
# A LIST IS THE BUG. Until now the compact/clear branch named its files one at
+12 -1
View File
@@ -192,6 +192,10 @@ mkdir -p "$state_dir" 2>/dev/null || true
# A FOURTH channel (milestone 307): standing RULES the write resembles. Its own
# file for the same reason as the others — a rule named once should not be
# re-offered on every subsequent write in the session.
#
# A FIFTH (milestone 444): the Systems whose RULINGS were shown in full. SHARED
# with scribe_tool_rules.sh, like the rule file: an area's rulings shown before
# a command must not be repeated in full before the next edit there.
idfile=""
syncfile=""
derivefile=""
@@ -200,6 +204,8 @@ exclude_q=""
sync_exclude_q=""
derive_exclude_q=""
rule_exclude_q=""
rulingsfile=""
rulings_q=""
if [ -n "$session_id" ]; then
safe_sid=$(printf '%s' "$session_id" | tr -c 'A-Za-z0-9._-' '_')
# What this write defined, for the end-of-turn question (milestone 439). A
@@ -214,6 +220,8 @@ if [ -n "$session_id" ]; then
syncfile="$state_dir/${safe_sid}.sync.ids"
derivefile="$state_dir/${safe_sid}.derive.ids"
rulefile="$state_dir/${safe_sid}.rules.ids"
rulingsfile="$state_dir/${safe_sid}.rulings.ids"
rulings_q=$(scribe_rulings_query "$rulingsfile")
if [ -f "$idfile" ]; then
seen=$(tr '\n' ',' < "$idfile" 2>/dev/null | sed 's/,$//')
[ -n "$seen" ] && exclude_q="&exclude_ids=${seen}"
@@ -242,7 +250,7 @@ fi
reached=1
body=$(curl -fsS --max-time 5 \
-H "Authorization: Bearer ${token}" \
"${url%/}/api/plugin/prior-art?path=${path_enc}&code=${code_enc}${repo_q}${exclude_q}${sync_exclude_q}${derive_exclude_q}${rule_exclude_q}${shapes_q}" 2>/dev/null) || { body=""; reached=0; }
"${url%/}/api/plugin/prior-art?path=${path_enc}&code=${code_enc}${repo_q}${exclude_q}${sync_exclude_q}${derive_exclude_q}${rule_exclude_q}${rulings_q}${shapes_q}" 2>/dev/null) || { body=""; reached=0; }
unreached_context=""
if [ "$reached" = 1 ]; then
scribe_reached "$state_dir" "${safe_sid:-nosession}"
@@ -271,6 +279,9 @@ if [ -n "$body" ]; then
if [ -n "$derivefile" ]; then
scribe_json_list "$body_flat" '.derive_keys' >> "$derivefile" || true
fi
if [ -n "$rulingsfile" ]; then
scribe_json_list "$body_flat" '.ruling_system_ids' >> "$rulingsfile" || true
fi
fi
fi
+19 -1
View File
@@ -68,6 +68,15 @@ lookup_dir=${event_cwd:-${CLAUDE_PROJECT_DIR:-$PWD}}
scope=$(scribe_scope_query "$lookup_dir")
[ -n "$scope" ] && repo_q="&${scope}"
# Where the command runs, against the repo's root (milestone 444): the paths a
# command names are relative to its cwd or absolute, and a System's patterns
# are relative to the root. The server does the arithmetic; this sends both.
where_q=""
repo_root=$(git -C "$lookup_dir" rev-parse --show-toplevel 2>/dev/null || true)
if [ -n "$repo_root" ]; then
where_q="&root=$(printf '%s' "$repo_root" | scribe_urlenc)&cwd=$(printf '%s' "$lookup_dir" | scribe_urlenc)"
fi
# THE SHARED SESSION LEDGER, and the thing most worth getting right here.
#
# scribe_prior_art.sh keeps the rules it has already named in
@@ -82,6 +91,8 @@ state_dir="${TMPDIR:-/tmp}/scribe-priorart"
mkdir -p "$state_dir" 2>/dev/null || true
rulefile=""
rule_exclude_q=""
rulingsfile=""
rulings_q=""
if [ -n "$session_id" ]; then
safe_sid=$(printf '%s' "$session_id" | tr -c 'A-Za-z0-9._-' '_')
rulefile="$state_dir/${safe_sid}.rules.ids"
@@ -91,6 +102,10 @@ if [ -n "$session_id" ]; then
[ -n "$rule_seen" ] && rule_exclude_q="&exclude_rule_ids=${rule_seen}"
# What the session actually OPENED, as against what it was shown (#4100).
rule_exclude_q="${rule_exclude_q}$(scribe_held_query "$state_dir/${safe_sid}.opened.ids")"
# The Systems whose rulings were shown in full — shared with the write-path
# hook, for the reason the rule file is.
rulingsfile="$state_dir/${safe_sid}.rulings.ids"
rulings_q=$(scribe_rulings_query "$rulingsfile")
fi
# `|| exit 0` here, unlike the prior-art hook: there is no local arm whose
@@ -98,7 +113,7 @@ fi
# than silence. See the header.
body=$(curl -fsS --max-time 5 \
-H "Authorization: Bearer ${token}" \
"${url%/}/api/plugin/tool-rules?tool=${tool_enc}&command=${cmd_enc}${repo_q}${rule_exclude_q}" 2>/dev/null) || exit 0
"${url%/}/api/plugin/tool-rules?tool=${tool_enc}&command=${cmd_enc}${repo_q}${where_q}${rule_exclude_q}${rulings_q}" 2>/dev/null) || exit 0
body_flat=$(printf '%s' "$body" | scribe_json_flat)
context=$(scribe_json_pick "$body_flat" '.context')
@@ -111,6 +126,9 @@ context=$(scribe_json_pick "$body_flat" '.context')
if [ -n "$rulefile" ]; then
scribe_json_list "$body_flat" '.rule_ids' | scribe_rules_append "$rulefile"
fi
if [ -n "$rulingsfile" ]; then
scribe_json_list "$body_flat" '.ruling_system_ids' >> "$rulingsfile" || true
fi
# ── The pre-act checkpoint (#4214, milestone 419) ────────────────────────
#
@@ -96,6 +96,8 @@ one line each — the statement, who decided, when, and the record it came from:
The description arrives with every record filed under that System, so a ruling
there reaches each session working in the area without having to win a search.
Once the System names its files (`path_patterns`), its rulings also arrive with
the first command or edit in a session that touches them.
A quote inside a work log does not: it surfaces only when a query happens to
match it, and the passage that matches is usually the prose around it — often a
session's *reading* of the ruling rather than the ruling.
+4
View File
@@ -18,6 +18,7 @@ from scribe.services import canonical_systems as canonical_systems_svc
from scribe.services import milestones as milestones_svc
from scribe.services import notes as notes_svc
from scribe.services import systems as systems_svc
from scribe.services.system_usage import record_system_pulled
# Below this, a project is young enough that the mild "which area is this
# about?" question stays proportionate; at or above it, a zero-Systems project
@@ -293,6 +294,9 @@ async def get_system(system_id: int) -> dict:
system = await systems_svc.get_system(uid, system_id)
if system is None:
raise ValueError(f"system {system_id} not found")
# The pull half of the rulings arm's measurement (milestone 444): a
# System whose rulings were shown, then opened.
record_system_pulled(user_id=uid, system_id=system_id, source="mcp_get_system")
records = await systems_svc.list_records_for_system(uid, system_id)
titles = await milestones_svc.titles_for({r.milestone_id for r in records})
issues, tasks, notes = [], [], []
+1
View File
@@ -60,6 +60,7 @@ from scribe.models.retrieval_log import RetrievalLog # noqa: E402, F401
from scribe.models.retrieval_tuning import RetrievalTuningEvent # noqa: E402, F401
from scribe.models.note_usage import NoteUsageEvent # noqa: E402, F401
from scribe.models.rule_usage import RuleUsageEvent # noqa: E402, F401
from scribe.models.system_usage import SystemUsageEvent # noqa: E402, F401
from scribe.models.project import Project # noqa: E402, F401
from scribe.models.milestone import Milestone # noqa: E402, F401
from scribe.models.task_log import TaskLog # noqa: E402, F401
+56
View File
@@ -0,0 +1,56 @@
from sqlalchemy import BigInteger, Index, Text
from sqlalchemy.orm import Mapped, mapped_column
from scribe.models import Base
from scribe.models.base import CreatedAtMixin, iso
SURFACED = "surfaced"
PULLED = "pulled"
class SystemUsageEvent(Base, CreatedAtMixin):
"""One row per time a System's rulings were SHOWN to the agent because its
files were touched, or the System was PULLED in full (milestone 444).
The third of the usage tables, after `note_usage_events` and
`rule_usage_events`, and separate from both for the reason the rule twin
gives: identity at restore. A System id is its own namespace, mapped
through the restore's system map; parked in either sibling's id column it
would come back attached to whatever note or rule took that number.
What it answers: whether an area's rulings, delivered by path rather than
by a ranker, are then read (`get_system`) — and so whether delivery by
path earns its line.
FK-free on `system_id`, `user_id` and `project_id`, like its siblings:
telemetry outlives the row it describes.
"""
__tablename__ = "system_usage_events"
id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
user_id: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
system_id: Mapped[int] = mapped_column(BigInteger, nullable=False)
# 'surfaced' | 'pulled'. Plain Text, no CHECK, like the siblings.
event: Mapped[str] = mapped_column(Text, nullable=False)
# Which surface produced it — a convention, not a vocabulary (see the
# note twin). `grep -rn record_system_ src/` is the authoritative list.
source: Mapped[str] = mapped_column(Text, nullable=False)
# The project the READER was in, not the System's own.
project_id: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
__table_args__ = (
Index("ix_system_usage_system_event", "system_id", "event"),
Index("ix_system_usage_created_at", "created_at"),
)
def to_dict(self) -> dict:
return {
"id": self.id,
"created_at": iso(self.created_at),
"user_id": self.user_id,
"system_id": self.system_id,
"event": self.event,
"source": self.source,
"project_id": self.project_id,
}
+19 -1
View File
@@ -194,8 +194,18 @@ async def pre_tool_rules():
different claims about the reader's
context, so they get different lines
(#4100).
root, cwd (opt) — the repo's absolute root and the command's
working directory. They turn the paths the
command names into repo-relative ones, for
the rulings arm (milestone 444).
seen_ruling_systems (opt) — comma-separated System ids whose rulings
were already shown this session, by either
arm; those get a one-line reference. SHARED
with /prior-art, like exclude_rule_ids.
Returns `context`, `rule_ids`, and `checkpoint` (#4214, milestone 419).
Returns `context`, `rule_ids`, and `checkpoint` (#4214, milestone 419),
plus `ruling_system_ids` — the Systems whose rulings were shown in full on
this call — when there were any.
`checkpoint` IS THE ONE PART OF THIS RESPONSE THAT IS NOT A HINT. It is
empty on almost every call. When present it carries `rule_id`, `title`,
@@ -221,6 +231,9 @@ async def pre_tool_rules():
result = await plugin_ctx_svc.build_tool_rule_hint(
g.user.id, tool, command,
project_id=project_id, exclude_rule_ids=exclude_rule_ids, held_rule_ids=held_rule_ids,
root=(request.args.get("root") or "").strip(),
cwd=(request.args.get("cwd") or "").strip(),
seen_ruling_systems=_int_list(request.args.get("seen_ruling_systems")),
)
return jsonify(result)
@@ -265,6 +278,10 @@ async def write_path_prior_art():
or `canon:<snippet_id>`) already named this
session by the ledger arm (#2900); its own
channel, like the two above.
seen_ruling_systems (opt) — System ids whose rulings were already shown
this session (milestone 444); shared with
/tool-rules. The response's `ruling_system_ids`
are the ones shown in full on this call.
(Returns a `checkpoint` block on the same contract as /tool-rules —
see that endpoint. The write-path HOOK deliberately does not act on
it: scribe_prior_art.sh carries a tested property that it never
@@ -306,6 +323,7 @@ async def write_path_prior_art():
repo_key=repo_bindings_svc.normalize_repo_key(repo) if repo else "",
exclude_derive=exclude_derive,
exclude_rule_ids=exclude_rule_ids, held_rule_ids=held_rule_ids,
seen_ruling_systems=_int_list(request.args.get("seen_ruling_systems")),
)
return jsonify(result)
+58 -1
View File
@@ -13,6 +13,7 @@ from scribe.models.rule_version import RuleVersion
from scribe.models.design_system import DesignSystem, DesignToken
from scribe.models.note_usage import NoteUsageEvent
from scribe.models.rule_usage import RuleUsageEvent
from scribe.models.system_usage import SystemUsageEvent
from scribe.models.retrieval_tuning import RetrievalTuningEvent
from scribe.models.canonical_system import CanonicalSystem
from scribe.models.rulebook import RuleRelation, rule_systems as rule_systems_t
@@ -88,8 +89,12 @@ logger = logging.getLogger(__name__)
# v19 (2026-10) added lesson_no_rule (#4631): the "no rule fits" answer and its
# reason. Without it a restored lesson that was judged to stand alone reads as
# never judged, and lands back on the unjudged list.
# v20 (2026-10) added systems.path_patterns and system_usage_events (milestone
# 444): the files that are each area, and whether an area's rulings were read
# once shown. The usage rows restore through the SYSTEM map, for the reason
# the rule twin restores through the rule map.
# Bump when the serialized schema changes.
BACKUP_VERSION = 19
BACKUP_VERSION = 20
# Every table this backup carries, by its REAL name. Paired with _NOT_INCLUDED
# below, these two lists must together account for the entire schema — which is
@@ -131,6 +136,9 @@ _BACKED_UP = [
"lesson_rule_links",
# v19 (2026-10): "no rule fits" answers (#4631).
"lesson_no_rule",
# v20 (2026-10): System usage telemetry (milestone 444), for the reason
# its note and rule twins travel.
"system_usage_events",
]
# Tables intentionally NOT in the backup, surfaced in the payload so the gap is
@@ -231,6 +239,7 @@ _COLUMN_EXCLUSIONS: dict[str, set[str]] = {
"note_usage_events": {"id"},
# Same as the note twin: the surrogate key is re-issued on insert.
"rule_usage_events": {"id"},
"system_usage_events": {"id"},
# Same again — and everything else travels, because each remaining column
# is part of the argument: what moved, from what, to what, by whom, why.
"retrieval_tuning_events": {"id"},
@@ -319,6 +328,7 @@ _IMPORT_COLUMN_EXCLUSIONS: dict[str, set[str]] = {
"lesson_no_rule": set(),
"note_usage_events": {"id"},
"rule_usage_events": {"id"},
"system_usage_events": {"id"},
"retrieval_tuning_events": {"id"},
"design_systems": {
"id", "deleted_at", "deleted_batch_id", "created_at", "updated_at",
@@ -444,6 +454,17 @@ def _usage_event_rows(rows) -> list[dict]:
]
def _system_usage_event_rows(rows) -> list[dict]:
return [
{
"user_id": r.user_id, "system_id": r.system_id, "event": r.event,
"source": r.source, "project_id": r.project_id,
"created_at": r.created_at.isoformat() if r.created_at else None,
}
for r in rows
]
def _rule_usage_event_rows(rows) -> list[dict]:
return [
{
@@ -802,6 +823,9 @@ async def export_full_backup() -> dict:
rule_usage_events = (
await session.execute(select(RuleUsageEvent))
).scalars().all()
system_usage_events = (
await session.execute(select(SystemUsageEvent))
).scalars().all()
# Oldest first, so a restored history reads in the order the dials
# actually moved — the sequence IS the argument when a surface has been
# walked up and down.
@@ -854,6 +878,7 @@ async def export_full_backup() -> dict:
"design_tokens": _design_token_rows(design_tokens),
"note_usage_events": _usage_event_rows(usage_events),
"rule_usage_events": _rule_usage_event_rows(rule_usage_events),
"system_usage_events": _system_usage_event_rows(system_usage_events),
"retrieval_tuning_events": _retrieval_tuning_event_rows(
retrieval_tuning_events
),
@@ -991,6 +1016,12 @@ async def export_user_backup(user_id: int) -> dict:
rule_usage_events = (await session.execute(
select(RuleUsageEvent).where(RuleUsageEvent.rule_id.in_(_rule_ids))
)).scalars().all() if _rule_ids else []
# Scoped through the SYSTEM, for the reason the rule usage events
# above are scoped through the rule: `user_id` is who it fired for.
_system_ids = [sy.id for sy in systems]
system_usage_events = (await session.execute(
select(SystemUsageEvent).where(SystemUsageEvent.system_id.in_(_system_ids))
)).scalars().all() if _system_ids else []
# Scoped on user_id, and here that IS the right column — unlike the
# rule usage events directly above. These record changes to this user's
# OWN retrieval settings, which is what `user_id` means on this table;
@@ -1054,6 +1085,7 @@ async def export_user_backup(user_id: int) -> dict:
"design_tokens": _design_token_rows(design_tokens),
"note_usage_events": _usage_event_rows(usage_events),
"rule_usage_events": _rule_usage_event_rows(rule_usage_events),
"system_usage_events": _system_usage_event_rows(system_usage_events),
"retrieval_tuning_events": _retrieval_tuning_event_rows(
retrieval_tuning_events
),
@@ -1566,6 +1598,22 @@ def _build_rule_usage_event(row: dict, maps: _Maps) -> RuleUsageEvent | None:
)
def _build_system_usage_event(row: dict, maps: _Maps) -> SystemUsageEvent | None:
"""Resolved through the SYSTEM map — see `_build_rule_usage_event`."""
sid = maps.systems.get(row.get("system_id", 0))
if sid is None:
return None
return SystemUsageEvent(
user_id=maps.users.get(row.get("user_id") or 0),
system_id=sid,
event=row.get("event", ""),
source=row.get("source", ""),
project_id=(maps.projects.get(row["project_id"])
if row.get("project_id") else None),
created_at=_dt(row.get("created_at")),
)
def _build_repo_binding(row: dict, maps: _Maps) -> RepoBinding | None:
"""Small, but losing these means every bound repo quietly stops loading
its project at session start."""
@@ -1816,6 +1864,7 @@ async def _restore_v2(data: dict) -> dict:
"settings": 0, "rulebooks": 0, "rulebook_topics": 0, "rules": 0,
"systems": 0, "record_systems": 0, "design_systems": 0,
"design_tokens": 0, "note_usage_events": 0, "rule_usage_events": 0,
"system_usage_events": 0,
"repo_bindings": 0,
"note_supersessions": 0, "code_shapes": 0, "code_shape_events": 0,
"code_shape_uses": 0, "canonical_systems": 0,
@@ -2105,6 +2154,14 @@ async def _restore_v2(data: dict) -> dict:
session.add(event)
stats["rule_usage_events"] += 1
# And the System twin, after the Systems (step 16 fills their map).
for ev in data.get("system_usage_events", []):
event = _build_system_usage_event(ev, maps)
if event is None:
continue
session.add(event)
stats["system_usage_events"] += 1
# 20. Repo bindings
for rb_data in data.get("repo_bindings", []):
binding = _build_repo_binding(rb_data, maps)
+51 -6
View File
@@ -48,6 +48,7 @@ from scribe.services.retrieval_telemetry import record_retrieval
from scribe.services.settings import get_setting
from scribe.services.systems import system_names_for
from scribe.services.text import elide
from scribe.services import system_rulings as system_rulings_svc
logger = logging.getLogger(__name__)
@@ -2187,6 +2188,7 @@ async def build_write_path_hint(
exclude_derive: list[str] | None = None,
exclude_rule_ids: list[int] | None = None,
held_rule_ids: list[int] | None = None,
seen_ruling_systems: list[int] | None = None,
) -> dict:
"""Prior-art hint for the plugin's PreToolUse hook on Write/Edit.
@@ -2253,7 +2255,7 @@ async def build_write_path_hint(
cfg = await get_writepath_config(user_id)
empty = {"context": "", "note_ids": [], "sync_note_ids": [], "config": cfg,
"suggested": [], "divergence": [], "derive": [], "derive_keys": [],
"rule_ids": []}
"rule_ids": [], "ruling_system_ids": []}
path = (path or "").strip()
if not cfg["enabled"] or not path:
return empty
@@ -2590,9 +2592,21 @@ async def build_write_path_hint(
design_text, design_dedup = await _design_arm(
user_id, project_id, path, set(exclude_derive or []),
)
# The rulings arm (milestone 444), decided here for the design arm's
# reasons: a lookup by path, so a write that matched no prior art still
# carries its area's rulings, and it never switches the ranked arms on.
rulings = await system_rulings_svc.rulings_for_paths(
user_id, project_id, [path], seen=seen_ruling_systems,
source="rulings_write_path",
)
if not staleness and not synced and not menu and not suggested and not divergence and not derive:
if design_text:
return {**empty, "context": design_text, "derive_keys": [design_dedup]}
if design_text or rulings["lines"]:
return {
**empty,
"context": "\n".join(rulings["lines"] + ([design_text] if design_text else [])),
"derive_keys": [design_dedup] if design_dedup else [],
"ruling_system_ids": rulings["system_ids"],
}
return empty
owners = await owner_names_for({
@@ -2614,7 +2628,9 @@ async def build_write_path_hint(
# Seeded with the staleness line, which is decided above the early
# return and so cannot wait for this list to exist.
lines: list[str] = list(staleness)
# First after staleness: it BINDS, where everything below is prior art.
# First after staleness: rulings and the design system BIND, where
# everything below is prior art.
lines.extend(rulings["lines"])
if design_text:
lines.append(design_text)
sync_note_ids: list[int] = []
@@ -2890,6 +2906,7 @@ async def build_write_path_hint(
),
"rule_ids": rule_ids,
"checkpoint": checkpoint,
"ruling_system_ids": rulings["system_ids"],
}
@@ -2901,18 +2918,46 @@ async def build_tool_rule_hint(
project_id: int = 0,
exclude_rule_ids: list[int] | None = None,
held_rule_ids: list[int] | None = None,
root: str = "",
cwd: str = "",
seen_ruling_systems: list[int] | None = None,
) -> dict:
"""Standing rules that may apply to the action about to be taken — matched
directly (`_tool_rule_hint`, where the design is written), then reached
through a linked lesson (`_rules_via_lessons`, #4633)."""
through a linked lesson (`_rules_via_lessons`, #4633) — and the rulings of
any area whose files the command names (milestone 444).
`root` and `cwd` are the repo's absolute root and the command's working
directory, from the hook; they are what turn the paths a command names
into the repo-relative paths a System's patterns are written in."""
out = await _tool_rule_hint(
user_id, tool_name, command, project_id=project_id,
exclude_rule_ids=exclude_rule_ids, held_rule_ids=held_rule_ids,
)
return await _add_rules_via_lessons(
out = await _add_rules_via_lessons(
user_id, out, project_id=project_id, exclude_rule_ids=exclude_rule_ids,
held_rule_ids=held_rule_ids, where=f"to this {tool_name} call",
)
# Its own arm, not part of the rule search: a lookup by path, with no
# floor and no budget, so it adds to the rule lines rather than competing
# with them. First, because a ruling is the operator's own decision.
# `ruling_system_ids` is set only when a ruling was shown: this arm fires
# on every command, and the hook reads an absent list as an empty one.
try:
paths = system_rulings_svc.command_paths(command, root=root, cwd=cwd) if project_id else []
if paths and (await get_writepath_config(user_id)).get("enabled"):
rulings = await system_rulings_svc.rulings_for_paths(
user_id, project_id, paths,
seen=seen_ruling_systems, source="rulings_pre_tool",
)
if rulings["lines"]:
out["context"] = "\n".join(
rulings["lines"] + ([out["context"]] if out.get("context") else [])
)
out["ruling_system_ids"] = rulings["system_ids"]
except Exception:
logger.debug("pre-tool rulings arm failed", exc_info=True)
return out
async def _tool_rule_hint(
+196
View File
@@ -0,0 +1,196 @@
"""An area's rulings, delivered when its files are touched (milestone 444).
A RULING is the operator's decision about how one area of the work must behave.
It lives in a `Rulings` section at the end of the area's System description
(writing-records.md says how one is written). The description already rides
every record filed under the System; this module is the other half: the first
command or edit in a session that touches the System's files
(`System.path_patterns`) shows its rulings, one line per System.
A LOOKUP, NOT A SEARCH. A shell command scores against prose decisions at noise
level, so nothing here is ranked: a path either falls under a System's
patterns or it does not. It takes no budget and no floor from the retrieval
arms beside it, and writes no retrieval_logs row — there is no score
distribution for it to join. Its surfacings go to `system_usage_events`.
ONCE IN FULL, THEN A REFERENCE (the #3750 shape). The hook keeps the Systems
already shown this session and passes them back; a repeat renders as one short
line rather than the rulings again, and is not counted as a surfacing.
Fails open everywhere: a delivery aid must never break the act it rides on.
"""
from __future__ import annotations
import logging
import posixpath
import re
import shlex
from scribe.services import systems as systems_svc
from scribe.services.system_usage import record_system_surfaced
from scribe.services.text import elide
logger = logging.getLogger(__name__)
# The section heading: `Rulings`, optionally as a markdown heading, bold, or
# with a trailing colon. The LAST one wins — the section sits at the end.
_HEADING = re.compile(r"^[ \t]*(?:#{1,6}[ \t]*)?\**Rulings\**[ \t]*:?[ \t]*$", re.I | re.M)
_BULLET = re.compile(r"^[ \t]*[-*][ \t]+(.*)$")
# Per line, so one System with a long list cannot crowd the others out.
RULING_CHARS = 300
RULINGS_PER_SYSTEM = 8
# A command is not a file list: these bound how much of it is read as paths.
_MAX_PATHS = 40
_LINE_SUFFIX = re.compile(r":\d+(?::\d+)?$")
_HAS_EXTENSION = re.compile(r"\.[A-Za-z0-9]{1,10}$")
_ASSIGNMENT = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*=")
def parse_rulings(description: str | None) -> list[str]:
"""The rulings in a System description, one string per ruling.
A bullet under the last `Rulings` heading is a ruling; an indented line
after one continues it. The section ends at the first line that is
neither — a heading or paragraph written after it is not a ruling.
"""
text = description or ""
found = list(_HEADING.finditer(text))
if not found:
return []
out: list[str] = []
for line in text[found[-1].end():].splitlines():
if not line.strip():
continue
bullet = _BULLET.match(line)
if bullet:
if bullet.group(1).strip():
out.append(bullet.group(1).strip())
elif out and line[:1] in (" ", "\t"):
out[-1] = f"{out[-1]} {line.strip()}"
else:
break
return out
def _relative(token: str, root: str, cwd_rel: str) -> str:
"""`token` as a repo-relative path, or "" when it is not one."""
if token.startswith("/"):
if not root or not token.startswith(root.rstrip("/") + "/"):
return ""
token = token[len(root.rstrip("/")) + 1:]
elif cwd_rel:
token = f"{cwd_rel}/{token}"
path = posixpath.normpath(token)
if path in (".", "") or path == ".." or path.startswith("../"):
return ""
return path
def command_paths(command: str, *, root: str = "", cwd: str = "") -> list[str]:
"""The repo-relative paths a shell command names, in order, deduplicated.
Reads and writes alike: a session forms its picture of an area by reading
it, which is exactly when that area's rulings should reach it. A token is
taken as a path when it has a `/` or a file extension; flags, URLs and
globs are not paths. `root` is the repo's absolute root and `cwd` the
command's working directory, both from the hook — an absolute path
outside the root, or a relative one that climbs out of it, is dropped.
"""
command = command or ""
try:
tokens = shlex.split(command, comments=False, posix=True)
except ValueError:
tokens = command.split()
root = (root or "").rstrip("/")
cwd_rel = ""
if root and cwd:
cwd = cwd.rstrip("/")
if cwd.startswith(root + "/"):
cwd_rel = cwd[len(root) + 1:]
out: list[str] = []
for raw in tokens:
token = raw.strip().strip("'\"").rstrip(";,)|&")
if token.startswith("-"):
if "=" not in token:
continue
token = token.split("=", 1)[1]
elif _ASSIGNMENT.match(token):
token = token.split("=", 1)[1]
if not token or "://" in token or any(c in token for c in "*?[]{}$`<>"):
continue
# `file.py:120` and a test id `file.py::test_x` name the file.
token = _LINE_SUFFIX.sub("", token.split("::", 1)[0])
if "/" not in token and not _HAS_EXTENSION.search(token):
continue
path = _relative(token, root, cwd_rel)
if path and path not in out:
out.append(path)
if len(out) >= _MAX_PATHS:
break
return out
def _full_line(system, path: str, rulings: list[str]) -> str:
shown = [elide(r, RULING_CHARS)[0] for r in rulings[:RULINGS_PER_SYSTEM]]
more = len(rulings) - len(shown)
listed = " ".join(f"({i}) {r}" for i, r in enumerate(shown, 1))
if more > 0:
listed += f" (+{more} more in `get_system({system.id})`)"
return (
f"> Rulings for `{path}` — the operator's decisions about "
f"{system.name} (System {system.id}): {listed} "
"Work here keeps to them; something that would depart from one is a "
"question for the operator, not an option to choose. "
"(Shown once per session.)"
)
def _reference_line(system, path: str) -> str:
return (
f"> `{path}` is in {system.name} (System {system.id}) — its rulings "
"were shown earlier this session and still apply."
)
async def rulings_for_paths(
user_id: int,
project_id: int,
paths: list[str],
*,
seen: list[int] | set[int] | None = None,
source: str,
) -> dict:
"""Rulings lines for the Systems whose files `paths` touch.
Returns {"lines": [...], "system_ids": [...]} — `system_ids` are the
Systems shown IN FULL on this call, for the hook to add to the session's
seen list and for the usage ledger; a System in `seen` gets a reference
line and is in neither. A System with patterns but no rulings says
nothing: there is no decision to deliver, and the charter already rides
its records.
"""
out: dict = {"lines": [], "system_ids": []}
if not project_id or not paths:
return out
try:
already = {int(s) for s in (seen or [])}
for system, hit in await systems_svc.systems_for_paths(user_id, project_id, paths):
rulings = parse_rulings(system.description)
if not rulings:
continue
if system.id in already:
out["lines"].append(_reference_line(system, hit[0]))
else:
out["lines"].append(_full_line(system, hit[0], rulings))
out["system_ids"].append(system.id)
if out["system_ids"]:
record_system_surfaced(
user_id=user_id, system_ids=out["system_ids"], source=source,
project_id=project_id,
)
except Exception:
logger.debug("rulings arm failed", exc_info=True)
return {"lines": [], "system_ids": []}
return out
+67
View File
@@ -0,0 +1,67 @@
"""System usage telemetry — were an area's rulings read once shown?
The twin of `note_usage` and `rule_usage` for Systems (milestone 444). The
rulings arm shows a System's rulings when a command or edit touches the
System's files; a pull is somebody then opening the System (`get_system`). The
ratio says whether delivering rulings by path earns its line.
Fire-and-forget like its siblings: telemetry never adds latency to, or
breaks, the surface it observes, and failures report through the shared
canary rather than vanishing.
"""
from __future__ import annotations
import logging
from scribe.models import async_session
from scribe.models.system_usage import PULLED, SURFACED, SystemUsageEvent
from scribe.services.background import report_telemetry_failure, spawn
logger = logging.getLogger(__name__)
async def _insert_events(rows: list[dict]) -> None:
"""Persist usage rows. Best-effort: failures degrade, visibly."""
try:
async with async_session() as session:
session.add_all([SystemUsageEvent(**row) for row in rows])
await session.commit()
except Exception:
await report_telemetry_failure("system_usage", "write")
def _rows(user_id, system_ids, event: str, source: str, project_id) -> list[dict]:
pid = int(project_id or 0) or None
return [
{"user_id": user_id, "system_id": int(sid), "event": event,
"source": source, "project_id": pid}
for sid in system_ids
]
def record_system_surfaced(
*, user_id: int | None, system_ids: list[int], source: str,
project_id: int | None = None,
) -> None:
"""Fire-and-forget: these Systems' rulings were shown in full. A repeat
rendered as a short reference is not a surfacing and is not recorded."""
try:
rows = _rows(user_id, system_ids, SURFACED, source, project_id)
except Exception:
logger.debug("system usage payload build failed", exc_info=True)
return
if rows:
spawn(_insert_events(rows), site="system_usage_write")
def record_system_pulled(
*, user_id: int | None, system_id: int, source: str,
project_id: int | None = None,
) -> None:
"""Fire-and-forget: a System was opened in full."""
try:
rows = _rows(user_id, [system_id], PULLED, source, project_id)
except Exception:
logger.debug("system usage payload build failed", exc_info=True)
return
spawn(_insert_events(rows), site="system_usage_write")
+15
View File
@@ -116,6 +116,21 @@ def _no_system_labels():
yield
@pytest.fixture(autouse=True)
def _no_rulings_arm():
"""Stub the rulings arm both PreToolUse builders run (milestone 444).
Autouse for _no_system_labels' reason: every write-path and tool-rule
test with a project in scope reaches it, and its first act is reading the
project's Systems from Postgres. Stubbed to "no area's files touched".
tests/test_system_rulings.py binds the real function at import time,
before this patch runs.
"""
with patch("scribe.services.system_rulings.rulings_for_paths",
AsyncMock(return_value={"lines": [], "system_ids": []})):
yield
@pytest.fixture(autouse=True)
def _no_task_log_arm():
"""Stub the task-log read arm that get_task / list_tasks / get_milestone
+7 -2
View File
@@ -27,7 +27,7 @@ def test_backup_version_is_current():
(Named for the number it asserted until v10, which is exactly the drift a
name-carrying-a-value invites; it now says what it checks.)"""
assert backup.BACKUP_VERSION == 19
assert backup.BACKUP_VERSION == 20
def _exportable_note(**over):
@@ -139,6 +139,7 @@ def _column_guard_targets():
from scribe.models.note_supersession import NoteSupersession
from scribe.models.note_usage import NoteUsageEvent
from scribe.models.rule_usage import RuleUsageEvent
from scribe.models.system_usage import SystemUsageEvent
from scribe.models.retrieval_tuning import RetrievalTuningEvent
from scribe.models.note_version import NoteVersion
from scribe.models.rule_version import RuleVersion
@@ -172,6 +173,7 @@ def _column_guard_targets():
"lesson_no_rule": (LessonNoRule, backup._lesson_no_rule_rows),
"note_usage_events": (NoteUsageEvent, backup._usage_event_rows),
"rule_usage_events": (RuleUsageEvent, backup._rule_usage_event_rows),
"system_usage_events": (SystemUsageEvent, backup._system_usage_event_rows),
"retrieval_tuning_events": (
RetrievalTuningEvent, backup._retrieval_tuning_event_rows,
),
@@ -290,6 +292,7 @@ def _import_guard_targets():
"lesson_no_rule": backup._build_lesson_no_rule,
"note_usage_events": backup._build_usage_event,
"rule_usage_events": backup._build_rule_usage_event,
"system_usage_events": backup._build_system_usage_event,
"retrieval_tuning_events": backup._build_retrieval_tuning_event,
"design_systems": backup._build_design_system,
"design_tokens": backup._build_design_token,
@@ -543,7 +546,9 @@ async def test_export_full_backup_contains_every_declared_section():
# v18: which rule each lesson is an instance of.
"lesson_rule_links",
# v19: the lessons judged to fall under no rule.
"lesson_no_rule"):
"lesson_no_rule",
# v20: whether an area's rulings were read once shown.
"system_usage_events"):
assert key in out, f"missing export section: {key}"
assert out[key] == []
+241
View File
@@ -0,0 +1,241 @@
"""An area's rulings reach the work that touches its files (milestone 444, #4757).
These pin:
- WHAT A RULING IS, as read: a bullet under the last `Rulings` heading of a
System's description, wrapped lines joined, the section ending at the first
line that is neither.
- WHICH PATHS A COMMAND NAMES: reads and writes alike, relative to the repo
root whatever the command's cwd; flags, URLs and globs are not paths, and a
path outside the repo is nobody's area.
- THE ARM: a lookup, not a search. No rulings or no patterns says nothing; a
path in two areas hears both; a repeat is one short reference line and is
neither counted as a surfacing nor handed back as newly shown.
- THE WIRING: both PreToolUse builders carry it, ahead of what they rank.
"""
from __future__ import annotations
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from scribe.services import plugin_context as pc
from scribe.services import system_rulings as sr
# Bound at import, before conftest's autouse stub replaces the module attribute.
from scribe.services.system_rulings import rulings_for_paths as real_rulings_for_paths
from tests.helpers import writepath_cfg
DOWNLOADS = """Fetching and keeping downloads healthy: stalls, retries, the blocklist.
Rulings
- Failed work is retried until it succeeds; no attempt limit. (Operator, 2026-09-19, #1234)
- A blocklisted release falls off the list after a while
and may be tried again. (Operator, 2026-09-20, #1240)
"""
def _system(sid, name, description, patterns):
return SimpleNamespace(id=sid, name=name, description=description, path_patterns=patterns)
# ── reading the section ──────────────────────────────────────────────────
def test_rulings_are_the_bullets_under_the_heading_with_wrapped_lines_joined():
assert sr.parse_rulings(DOWNLOADS) == [
"Failed work is retried until it succeeds; no attempt limit. (Operator, 2026-09-19, #1234)",
"A blocklisted release falls off the list after a while and may be tried again. "
"(Operator, 2026-09-20, #1240)",
]
@pytest.mark.parametrize("heading", ["Rulings", "## Rulings", "**Rulings**", "Rulings:", "### rulings"])
def test_the_heading_may_be_written_several_ways(heading):
assert sr.parse_rulings(f"Charter.\n\n{heading}\n- One. (Operator)\n") == ["One. (Operator)"]
def test_no_section_and_an_empty_description_have_no_rulings():
assert sr.parse_rulings("A charter that mentions rulings in passing.") == []
assert sr.parse_rulings("") == []
assert sr.parse_rulings(None) == []
def test_the_section_ends_where_a_paragraph_starts():
text = "Rulings\n- Kept. (Operator)\n\nNotes written after the section.\n- Not a ruling.\n"
assert sr.parse_rulings(text) == ["Kept. (Operator)"]
# ── the paths a command names ────────────────────────────────────────────
ROOT = "/home/me/repo"
@pytest.mark.parametrize("command,cwd,expected", [
("sed -n 1,80p src/app/downloads.py", ROOT, ["src/app/downloads.py"]),
# Absolute inside the root, and outside it.
(f"cat {ROOT}/src/app/x.py /etc/hosts", ROOT, ["src/app/x.py"]),
# A command run from a subdirectory names paths relative to it.
("grep -n retry downloads.py", f"{ROOT}/src/app", ["src/app/downloads.py"]),
("cat ../../../elsewhere/x.py", f"{ROOT}/src", []),
# Flags, URLs and globs are not paths; a `--flag=path` and a `path:line` are.
("pytest -q --rootdir=tests/unit tests/test_x.py::test_y", ROOT,
["tests/unit", "tests/test_x.py"]),
("curl https://example.com/a/b.json", ROOT, []),
("ls src/*.py", ROOT, []),
("vim src/app/x.py:120", ROOT, ["src/app/x.py"]),
# Repeats collapse; a bare word with no slash or extension is not a path.
("git add src/a.py src/a.py && git commit -m done", ROOT, ["src/a.py"]),
])
def test_command_paths(command, cwd, expected):
assert sr.command_paths(command, root=ROOT, cwd=cwd) == expected
def test_without_a_root_relative_paths_are_taken_as_given_and_absolute_ones_dropped():
assert sr.command_paths("cat src/a.py /abs/b.py") == ["src/a.py"]
def test_an_unbalanced_quote_still_yields_paths():
assert sr.command_paths("echo 'oops src/a.py") == ["src/a.py"]
# ── the arm ──────────────────────────────────────────────────────────────
async def _arm(systems, paths, seen=None):
lister = AsyncMock(return_value=systems)
recorder = MagicMock()
with patch.object(sr.systems_svc, "list_systems", lister), \
patch.object(sr, "record_system_surfaced", recorder):
out = await real_rulings_for_paths(1, 2, paths, seen=seen, source="rulings_test")
return out, recorder
@pytest.mark.asyncio
async def test_a_touched_area_with_rulings_shows_them_once_in_full_and_records_it():
downloads = _system(104, "Download lifecycle", DOWNLOADS, ["src/app/downloads"])
out, recorder = await _arm([downloads], ["src/app/downloads/retry.py"])
[line] = out["lines"]
assert line.startswith("> Rulings for `src/app/downloads/retry.py`")
assert "Download lifecycle (System 104)" in line
assert "(1) Failed work is retried until it succeeds" in line
assert "(2) A blocklisted release falls off" in line
assert out["system_ids"] == [104]
recorder.assert_called_once()
assert recorder.call_args.kwargs["system_ids"] == [104]
assert recorder.call_args.kwargs["source"] == "rulings_test"
@pytest.mark.asyncio
async def test_an_area_without_rulings_says_nothing():
plain = _system(5, "Billing", "What billing is for, and nothing decided.", ["src/billing"])
out, recorder = await _arm([plain], ["src/billing/invoice.py"])
assert out == {"lines": [], "system_ids": []}
recorder.assert_not_called()
@pytest.mark.asyncio
async def test_an_area_without_patterns_claims_no_files():
unnamed = _system(104, "Download lifecycle", DOWNLOADS, [])
out, _ = await _arm([unnamed], ["src/app/downloads/retry.py"])
assert out["lines"] == []
@pytest.mark.asyncio
async def test_a_path_in_two_areas_hears_both():
downloads = _system(104, "Download lifecycle", DOWNLOADS, ["src/app/downloads"])
storage = _system(7, "Storage", "Where files go.\n\nRulings\n- Never delete a user file. (Operator)\n",
["src/app/**/*.py"])
out, _ = await _arm([downloads, storage], ["src/app/downloads/retry.py"])
assert len(out["lines"]) == 2
assert out["system_ids"] == [104, 7]
@pytest.mark.asyncio
async def test_a_repeat_is_a_reference_and_is_not_counted_again():
downloads = _system(104, "Download lifecycle", DOWNLOADS, ["src/app/downloads"])
out, recorder = await _arm([downloads], ["src/app/downloads/retry.py"], seen=[104])
[line] = out["lines"]
assert "shown earlier this session" in line
assert "Failed work is retried" not in line
assert out["system_ids"] == []
recorder.assert_not_called()
@pytest.mark.asyncio
async def test_no_project_or_no_paths_reads_nothing():
lister = AsyncMock(return_value=[])
with patch.object(sr.systems_svc, "list_systems", lister):
assert await real_rulings_for_paths(1, 0, ["src/a.py"], source="t") == {"lines": [], "system_ids": []}
assert await real_rulings_for_paths(1, 2, [], source="t") == {"lines": [], "system_ids": []}
lister.assert_not_called()
@pytest.mark.asyncio
async def test_a_failing_lookup_fails_open():
with patch.object(sr.systems_svc, "list_systems", AsyncMock(side_effect=RuntimeError("db"))):
out = await real_rulings_for_paths(1, 2, ["src/a.py"], source="t")
assert out == {"lines": [], "system_ids": []}
@pytest.mark.asyncio
async def test_a_long_list_is_bounded_per_area():
many = "Rulings\n" + "".join(f"- Ruling {i}. (Operator)\n" for i in range(12))
area = _system(9, "Busy", many, ["src"])
out, _ = await _arm([area], ["src/x.py"])
assert "(8) Ruling 7." in out["lines"][0]
assert "Ruling 8." not in out["lines"][0]
assert "+4 more in `get_system(9)`" in out["lines"][0]
# ── the wiring ───────────────────────────────────────────────────────────
RULING_LINE = "> Rulings for `src/a.py` — the operator's decisions about A (System 3): (1) x"
@pytest.mark.asyncio
async def test_the_tool_arm_leads_with_rulings_and_hands_back_what_it_showed():
rulings = AsyncMock(return_value={"lines": [RULING_LINE], "system_ids": [3]})
with patch.object(pc, "_tool_rule_hint", AsyncMock(return_value={
"context": "> a rule line", "rule_ids": [8], "checkpoint": {}})), \
patch.object(pc, "get_writepath_config", AsyncMock(return_value=writepath_cfg())), \
patch.object(pc.system_rulings_svc, "rulings_for_paths", rulings):
out = await pc.build_tool_rule_hint(
1, "Bash", f"cat {ROOT}/src/a.py", project_id=2,
root=ROOT, cwd=ROOT, seen_ruling_systems=[5],
)
assert out["context"].splitlines() == [RULING_LINE, "> a rule line"]
assert out["ruling_system_ids"] == [3]
args, kwargs = rulings.call_args
assert args[2] == ["src/a.py"]
assert kwargs["seen"] == [5] and kwargs["source"] == "rulings_pre_tool"
@pytest.mark.asyncio
async def test_the_tool_arm_adds_no_key_when_no_ruling_was_shown():
with patch.object(pc, "_tool_rule_hint", AsyncMock(return_value={
"context": "", "rule_ids": [], "checkpoint": {}})), \
patch.object(pc, "get_writepath_config", AsyncMock(return_value=writepath_cfg())):
out = await pc.build_tool_rule_hint(1, "Bash", "ls", project_id=2)
assert "ruling_system_ids" not in out
@pytest.mark.asyncio
async def test_a_write_with_no_prior_art_still_carries_its_areas_rulings():
rulings = AsyncMock(return_value={"lines": [RULING_LINE], "system_ids": [3]})
with patch.object(pc, "get_writepath_config", AsyncMock(return_value=writepath_cfg())), \
patch.object(pc.snippets_svc, "list_snippets", AsyncMock(return_value=([], 0))), \
patch.object(pc, "semantic_search_notes", AsyncMock(return_value=[])), \
patch.object(pc, "record_retrieval", MagicMock()), \
patch.object(pc.projects_svc, "get_project", AsyncMock(return_value=None)), \
patch.object(pc.system_rulings_svc, "rulings_for_paths", rulings):
out = await pc.build_write_path_hint(
1, "src/a.py", code="x = 1", project_id=2, seen_ruling_systems=[9],
)
assert out["context"] == RULING_LINE
assert out["ruling_system_ids"] == [3]
args, kwargs = rulings.call_args
assert args[2] == ["src/a.py"]
assert kwargs["seen"] == [9] and kwargs["source"] == "rulings_write_path"