dev → main: rule-arm review, moment usage by rule, no-subject auto-inject, copy check, S3 test wording #215

Merged
bvandeusen merged 15 commits from dev into main 2026-10-22 20:26:40 -04:00
4 changed files with 80 additions and 83 deletions
Showing only changes of commit 71260a28e2 - Show all commits
+59 -53
View File
@@ -1458,14 +1458,10 @@ async def build_write_path_hint(
logger.warning("write-time derive check failed", exc_info=True)
staleness: list[str] = []
# The standing-rule arm further down is deliberately left on the far side
# of the guard below: it runs a SEMANTIC search, and moving it above would
# run an embedding query on every write in the session.
# The design arm is decided HERE, above the guard, so a UI write
# that matched no prior art still carries it — and it returns on its own
# rather than joining the guard's condition, because joining it would let
# a design line switch the standing-rule arm below on for writes where it
# has never run, moving that arm's call distribution under its floor.
# rather than joining the guard's condition: a design line alone is not a
# reason to render the prior-art menu's frame.
design_text, design_dedup = await _design_arm(
user_id, project_id, path, set(exclude_derive or []),
)
@@ -1476,7 +1472,62 @@ async def build_write_path_hint(
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:
# ── Standing rules that may apply here ───────────────────────────────
#
# A SUGGESTION, not a binding surface, and the distinction is the design:
# a rule BINDS by being tagged to an area the project works in,
# resolved deterministically at enter_project. This arm reaches for
# something weaker and still useful — a conditional rule whose trigger
# resembles what is being written, noticed at the moment it is relevant
# rather than by being resident in every session.
#
# EVERY RULE IS ELIGIBLE, including one already named this session: a
# rule the session was told about an hour ago is not a rule in front of
# the reader now.
#
# The arm itself is `retrieval_pipeline.run_rule_arm`: the band, the
# fresh/repeat split, the unconditional call row and the fresh-only
# surfacing rows are written there once for all three
# rule arms. What is decided HERE is only what makes this moment this
# moment — the query is the code being written (or the path, for an edit
# with no body), and a line says the rule may apply "here".
#
# EVERY WRITE WITH A PATH GETS THE CHECK, whether or not prior art matched:
# a rule about the file being written does not depend on a snippet
# resembling it. It runs before the notes guard below for that reason,
# at the price of one more search per write; its query, `code or path`,
# differs from the notes arm's concept query, so it is its own embedding.
#
# Fails open like every other arm: a rule hint must never break a write.
# The direct band, then a rule reached through a linked lesson,
# composed by `_rule_moment` exactly as the other two rule builders are.
rule_lines: list[str] = []
rule_ids: list[int] = []
shown_rule_ids: list[int] = []
checkpoint: dict = {}
try:
result = await _rule_moment(
rp.WRITE_PATH_RULE,
rp.RuleMoment(
user_id=user_id, query=code or path, project_id=project_id,
where="here", checkpoint_where="here",
exclude=frozenset(exclude_rule_ids or []),
held=frozenset(held_rule_ids or []),
),
floor=cfg["rule_threshold"], budget=cfg["rule_top_k"],
checkpoint_floor=cfg["checkpoint_threshold"],
)
rule_lines = result.lines
rule_ids.extend(result.rule_ids)
# Every rule line the band showed, repeats included — what the
# soft-link recorder pairs with this response's lessons.
shown_rule_ids = result.shown_rule_ids
checkpoint = result.checkpoint
except Exception:
logger.debug("write-path rule arm failed", exc_info=True)
if (not staleness and not synced and not menu and not suggested
and not divergence and not derive and not rule_lines):
if design_text or rulings["lines"]:
return {
**empty,
@@ -1612,52 +1663,7 @@ async def build_write_path_hint(
user_id=user_id, note_ids=ids, source=arm, project_id=project_id,
)
# ── Standing rules that may apply here ───────────────────────────────
#
# A SUGGESTION, not a binding surface, and the distinction is the design:
# a rule BINDS by being tagged to an area the project works in,
# resolved deterministically at enter_project. This arm reaches for
# something weaker and still useful — a conditional rule whose trigger
# resembles what is being written, noticed at the moment it is relevant
# rather than by being resident in every session.
#
# EVERY RULE IS ELIGIBLE, including one already named this session: a
# rule the session was told about an hour ago is not a rule in front of
# the reader now.
#
# The arm itself is `retrieval_pipeline.run_rule_arm`: the band, the
# fresh/repeat split, the unconditional call row and the fresh-only
# surfacing rows are written there once for all three
# rule arms. What is decided HERE is only what makes this moment this
# moment — the query is the code being written (or the path, for an edit
# with no body), and a line says the rule may apply "here".
#
# Fails open like every other arm: a rule hint must never break a write.
# The direct band, then a rule reached through a linked lesson,
# composed by `_rule_moment` exactly as the other two rule builders are.
rule_ids: list[int] = []
shown_rule_ids: list[int] = []
checkpoint: dict = {}
try:
result = await _rule_moment(
rp.WRITE_PATH_RULE,
rp.RuleMoment(
user_id=user_id, query=code or path, project_id=project_id,
where="here", checkpoint_where="here",
exclude=frozenset(exclude_rule_ids or []),
held=frozenset(held_rule_ids or []),
),
floor=cfg["rule_threshold"], budget=cfg["rule_top_k"],
checkpoint_floor=cfg["checkpoint_threshold"],
)
lines.extend(result.lines)
rule_ids.extend(result.rule_ids)
# Every rule line the band showed, repeats included — what the
# soft-link recorder pairs with this response's lessons.
shown_rule_ids = result.shown_rule_ids
checkpoint = result.checkpoint
except Exception:
logger.debug("write-path rule arm failed", exc_info=True)
lines.extend(rule_lines)
# A lesson and a rule on this one response: the soft-link
# recorder counts the pair, keyed on the FILE — every edit to one file is
+9 -22
View File
@@ -33,16 +33,8 @@ pytestmark = pytest.mark.usefixtures("_bind_user")
# each of these does.
# The write-path hint returns early when a write matched nothing at all — no
# staleness, no synced record, no prior-art menu, no shape signal. The rule arm
# sits deliberately on the FAR side of that guard, because it runs a semantic
# search and moving it above would mean an embedding query on every write in
# the session (which is why its gating is a separate question from
# precision).
#
# So a fixture that stubs every other arm to empty never reaches the rule arm
# at all — which is what the first run of this file did. The note hit below is
# not decoration: it is the condition the arm requires in order to fire.
# A prior-art hit for the fixtures that want the menu beside the rule lines.
# The rule arm runs on every write with a path, whether or not this matched.
_PRIOR_ART = [(0.72, fake_note(id=9, title="debounce helper", user_id=1,
note_type="snippet"))]
@@ -180,21 +172,16 @@ async def test_the_arm_searches_on_its_OWN_bar_not_the_code_one():
@pytest.mark.asyncio
async def test_the_arm_does_not_fire_on_a_write_that_matched_nothing():
"""The gate, pinned — because the fixture above now depends on it and a
silent change would make every other test here pass vacuously.
A write matching no prior art returns before the rule arm runs. That is
deliberate: the arm is a semantic search, and ungating it means an
embedding query on every write in the session. The gate stays until the
arm's precision is fixed, so this failing is a reason to look at precision
rather than to update the assertion.
"""
async def test_a_write_that_matched_no_prior_art_still_gets_its_rule_check():
"""A rule about the file being written does not depend on a snippet
resembling it, so the arm runs before the prior-art guard, not behind it."""
rec = MagicMock()
hits = [(0.71, fake_rule(id=156, title="A wait with no deadline is a bug"))]
await _run_arm(hits, rec, prior_art=[])
out = await _run_arm(hits, rec, prior_art=[])
assert rec.call_count == 0
assert rec.call_args.kwargs["rule_ids"] == [156]
assert "get_rule(156)" in out["context"]
assert out["rule_ids"] == [156]
@pytest.mark.asyncio
+7 -6
View File
@@ -181,12 +181,13 @@ async def test_a_ui_write_with_no_prior_art_still_hears_the_design_system():
@pytest.mark.asyncio
async def test_a_design_only_write_does_not_switch_the_rule_arm_on():
"""The rule arm runs only where there is prior art. A design line that
joined that gate would start a semantic rule search on every UI write —
a new population of calls under a floor tuned without them."""
_, rules = await _hint(_quiet())
rules.assert_not_called()
async def test_a_write_with_only_a_design_line_still_gets_its_rule_check():
"""Every write with a path gets the rule check, prior art or not. With
nothing ranked, the design line still arrives on its own."""
out, rules = await _hint(_quiet())
rules.assert_awaited()
assert rules.await_args_list[0].args[1] == "frontend/src/B.vue"
assert out["context"].startswith("> Design system binds `frontend/src/B.vue`")
@pytest.mark.asyncio
+5 -2
View File
@@ -614,7 +614,9 @@ async def test_trivial_payload_skips_the_semantic_arm_entirely():
out = await pc.build_write_path_hint(1, "src/x.py", code="x = 1")
assert out["context"] == "" and out["note_ids"] == []
search.assert_not_called()
rec.assert_not_called()
# The rule arm still checks the write and logs its own row; the notes arm
# logs none.
assert [c for c in rec.call_args_list if c.kwargs["source"] == "write_path"] == []
@pytest.mark.asyncio
@@ -798,7 +800,8 @@ async def test_telemetry_records_the_query_actually_sent():
patch.object(pc, "semantic_search_notes", AsyncMock(return_value=[])), \
patch.object(pc, "record_retrieval", rec):
await pc.build_write_path_hint(1, "src/x.py", code=REAL_CODE)
logged = rec.call_args.kwargs["query"]
row = next(c for c in rec.call_args_list if c.kwargs["source"] == "write_path")
logged = row.kwargs["query"]
assert logged.startswith("debounce(fn, wait=0.25)")
assert "nonlocal timer" not in logged