From db2427a3dcb6cee2ae178166eea89e77277d0a59 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Thu, 22 Oct 2026 20:14:41 -0400 Subject: [PATCH] feat(6249): the write-path hint asks whether a new definition copies a close snippet The derive check matches by name, so a renamed copy of a recorded snippet passed it and showed only as "similar" prior art. When a write introduces a definition the file does not hold (shape_ledger.new_shapes_at) and an unseen snippet on the menu clears the write-path floor by WRITEPATH_COPY_MARGIN (0.80 at the default floor), the hint now says to reuse it or extract the shared shape before another copy. Names the derive check or the stamping feed already spoke for are left to them. Covers recorded snippets only: the ledger keeps no bodies to compare an unrecorded sibling against. Co-Authored-By: Claude Opus 5.5 --- src/grimoire/services/plugin_context.py | 49 +++++++++++++++++++++ src/grimoire/services/shape_ledger.py | 23 ++++++++++ tests/test_integration_shape_classify.py | 22 ++++++++++ tests/test_write_path_trigger.py | 54 ++++++++++++++++++++++++ 4 files changed, 148 insertions(+) diff --git a/src/grimoire/services/plugin_context.py b/src/grimoire/services/plugin_context.py index b111f7e3..cd153ebf 100644 --- a/src/grimoire/services/plugin_context.py +++ b/src/grimoire/services/plugin_context.py @@ -143,6 +143,11 @@ WRITEPATH_ENABLED_KEY = "kb_writepath_enabled" WRITEPATH_THRESHOLD_KEY = SURFACES["write_path"].floor_key WRITEPATH_DEFAULT_ENABLED = True WRITEPATH_DEFAULT_THRESHOLD = SURFACES["write_path"].floor_default +# How far past the write-path floor a snippet must resemble a NEW definition +# before the hint asks whether the code is a copy of it. A margin, not a second +# floor, so it moves when the operator tunes the write-path floor: at the +# default 0.68 the copy bar is 0.80. +WRITEPATH_COPY_MARGIN = 0.12 # The standing-rule arms get their own bar, for a third corpus: 0.68 above is # calibrated on code against note PROSE, not code against RULE TEXT. @@ -1444,6 +1449,9 @@ async def build_write_path_hint( note.title, note.note_type, note.data, note.body, ), "seen": int(note.id) in excluded, + # Read by the copy check, which asks for more + # resemblance than the menu does. + "score": float(score), # Carried, not re-read off the rendered marker. The # marker is prose assembled for a human and it already # varies by kind, language and the `seen` flag — a @@ -1498,6 +1506,29 @@ async def build_write_path_hint( derive = [d for d in found if d.get("key") not in skip] except Exception: logger.warning("write-time derive check failed", exc_info=True) + # The COPY check: the derive check matches by name, so a renamed copy + # of a recorded snippet passes it and shows only as "similar" prior art. + # When this write introduces a definition and an unseen snippet on the + # menu resembles it well past the menu's own floor, say that a copy is + # the moment to extract. Only a recorded snippet can be seen this way — + # the ledger keeps no bodies to compare an unrecorded sibling against. + copies: list[dict] = [] + copy_bar = cfg["threshold"] + WRITEPATH_COPY_MARGIN + strong = [ + it for _m, it in menu + if it.get("kind") == "snippet" and not it.get("seen") + and it.get("score", 0.0) >= copy_bar + ] + if strong and stamp_shapes and project_id: + named = {d["symbol"] for d in derive} | {s["symbol"] for s in suggested} + try: + fresh = await shape_ledger_svc.new_shapes_at(project_id, path, stamp_shapes) + copies = [ + {"symbol": n, "kind": k, "snippet": strong[0]} + for k, n in fresh if n not in named + ] + except Exception: + logger.warning("write-time copy check failed", exc_info=True) staleness: list[str] = [] # The design arm is decided HERE, above the guard, so a UI write @@ -1681,6 +1712,8 @@ async def build_write_path_hint( lines.append(_divergence_line(path, divergence)) if derive: lines.append(_derive_line(path, derive)) + if copies: + lines.append(_copy_line(copies)) # Split by arm, which is the whole reason this table exists. The place arm # carries no score and so has no home in retrieval_logs, yet it fires on @@ -2008,6 +2041,22 @@ def _divergence_line(path: str, divergence: list[dict]) -> str: ) +def _copy_line(copies: list[dict]) -> str: + """A new definition that closely resembles a recorded snippet under + another name: the derive check cannot see a renamed copy, so this says + it — as a question for the writer, since resemblance is not identity.""" + snip = copies[0]["snippet"] + names = ", ".join( + f"`{'.' if c['kind'] == 'css' else ''}{c['symbol']}`" for c in copies + ) + return ( + f"> {names} {'is' if len(copies) == 1 else 'are'} new in this file and " + f"this code resembles snippet #{snip['id']} `{snip.get('name') or snip['title']}` " + f"(similar {snip['score']:.2f}). If it copies that shape, reuse #{snip['id']} " + "or extract the shared shape now, before another copy is written." + ) + + def _suggest_line(path: str, suggested: list[dict]) -> str: """One line naming the recorded shape this code looks like, as evidence for the judgment the agent gives at the end of the turn — not a record of diff --git a/src/grimoire/services/shape_ledger.py b/src/grimoire/services/shape_ledger.py index 90598c15..475470c0 100644 --- a/src/grimoire/services/shape_ledger.py +++ b/src/grimoire/services/shape_ledger.py @@ -2706,6 +2706,29 @@ async def write_time_derive( return out +async def new_shapes_at( + project_id: int, path: str, shapes: list[tuple[str, str]] +) -> list[tuple[str, str]]: + """The (kind, name) shapes the hook named at ``path`` that the ledger has + no live row for there — definitions this write introduces, as opposed to + ones it edits inside. Order follows ``shapes``.""" + wanted = [(k, n) for k, n in shapes if n] + if not wanted: + return [] + async with async_session() as session: + rows = ( + await session.execute( + select(CodeShape.kind, CodeShape.symbol).where( + CodeShape.project_id == project_id, + CodeShape.path == path, + CodeShape.vanished_at.is_(None), + ) + ) + ).all() + here = {(k, _norm_symbol(s)) for k, s in rows} + return [(k, n) for k, n in wanted if (k, _norm_symbol(n)) not in here] + + async def flag_divergence(project_id: int, *, since: datetime | None) -> int: """Flag shapes created after ``since`` (the previous refresh) that sit where a canon dominates and were not proposed as that canon. With no diff --git a/tests/test_integration_shape_classify.py b/tests/test_integration_shape_classify.py index 2411be31..db5a4260 100644 --- a/tests/test_integration_shape_classify.py +++ b/tests/test_integration_shape_classify.py @@ -665,6 +665,28 @@ async def test_write_time_derive_names_the_family_or_the_canon_for_a_name(seeded assert await write_time_derive(pid, "v/B.vue", [("css", "log-empty")]) == [] +@pytest.mark.integration +async def test_new_shapes_at_names_only_the_definitions_the_file_does_not_hold(seeded): + """A shape the ledger holds at the path is an edit inside it; one it + does not hold there — even under a name living elsewhere — is new.""" + from grimoire.services.shape_ledger import new_shapes_at + + pid = seeded["pid"] + await sync_repo_shapes(pid, REPO, _defs( + ("src/a.py", "sym", "load", "def load():", "def load():\n return 1"), + ("src/b.py", "sym", "save", "def save():", "def save():\n return 2"), + ("v/A.vue", "css", "log-empty", ".log-empty {", ".log-empty { color: red }"), + ), seen_marker="m1") + + assert await new_shapes_at(pid, "src/a.py", [("sym", "load"), ("sym", "save")]) == [ + ("sym", "save"), + ] + # CSS compares without the leading dot, as the rest of the ledger does. + assert await new_shapes_at(pid, "v/A.vue", [("css", ".log-empty")]) == [] + assert await new_shapes_at(pid, "src/new.py", [("sym", "load")]) == [("sym", "load")] + assert await new_shapes_at(pid, "src/a.py", [("sym", "")]) == [] + + @pytest.mark.integration async def test_derive_new_names_the_copy_that_joined_a_family_since_the_stamp(seeded): """The first sync seeds one `slug`; a later sync adds an identical diff --git a/tests/test_write_path_trigger.py b/tests/test_write_path_trigger.py index 0be0c209..93967598 100644 --- a/tests/test_write_path_trigger.py +++ b/tests/test_write_path_trigger.py @@ -1580,6 +1580,60 @@ async def test_the_write_time_derive_check_names_a_family_or_a_canon_in_band(): assert out["context"] == "" and out["derive"] == [] +@pytest.mark.asyncio +async def test_a_new_definition_close_to_a_snippet_is_asked_whether_it_is_a_copy(): + """The derive check matches by name, so a renamed copy of a recorded + snippet passes it. A new definition whose code resembles an unseen + snippet past the copy bar gets the extraction prompt; a shape the file + already holds, a weaker hit, a seen hit, or a name the derive check + already named does not.""" + from grimoire.services import plugin_context as pc + note = fake_note(id=31, title="debounce — rate-limit a callback", user_id=1, + note_type="snippet") + + async def run(score, fresh, *, derive=(), exclude=None): + new_at = AsyncMock(return_value=fresh) + with patch.multiple( + pc, get_writepath_config=AsyncMock(return_value=_cfg()), + semantic_search_notes=AsyncMock(return_value=[(score, note)]), + record_retrieval=MagicMock(), owner_names_for=AsyncMock(return_value={}), + ), patch.object(pc.snippets_svc, "list_snippets", AsyncMock(return_value=([], 0))), \ + patch.object(pc.shape_ledger_svc, "recent_pulls", AsyncMock(return_value={})), \ + patch.object(pc.shape_ledger_svc, "write_time_divergence", AsyncMock(return_value=[])), \ + patch.object(pc.shape_ledger_svc, "write_time_derive", + AsyncMock(return_value=list(derive))), \ + patch.object(pc.shape_ledger_svc, "new_shapes_at", new_at): + out = await pc.build_write_path_hint( + 1, "src/x.py", code=REAL_CODE, project_id=24, + stamp_shapes=[("sym", "throttle_calls")], exclude_ids=exclude, + ) + return out["context"], new_at + + bar = pc.WRITEPATH_DEFAULT_THRESHOLD + pc.WRITEPATH_COPY_MARGIN + ctx, new_at = await run(bar + 0.04, [("sym", "throttle_calls")]) + new_at.assert_awaited_once_with(24, "src/x.py", [("sym", "throttle_calls")]) + assert "> `throttle_calls` is new in this file and this code resembles snippet #31 " \ + "`debounce`" in ctx + assert f"(similar {bar + 0.04:.2f}). If it copies that shape, reuse #31" in ctx + + # The file already holds the shape: an edit inside it, not a copy. + ctx, _ = await run(bar + 0.04, []) + assert "is new in this file" not in ctx + # On the menu, but under the copy bar: no ledger read at all. + ctx, new_at = await run(bar - 0.04, [("sym", "throttle_calls")]) + assert "is new in this file" not in ctx and "similar" in ctx + new_at.assert_not_awaited() + # Shown already this session: the line was said with its first showing. + ctx, _ = await run(bar + 0.04, [("sym", "throttle_calls")], exclude=[31]) + assert "is new in this file" not in ctx + # The derive check named the shape: its line is the stronger claim. + family = {"symbol": "throttle_calls", "kind": "sym", "key": "dup:1", + "family": {"group": "dup:1", "label": "throttle_calls", "identical": True, + "files": ["a/y.py"], "file_count": 1, "size": 2}} + ctx, _ = await run(bar + 0.04, [("sym", "throttle_calls")], derive=[family]) + assert "is new in this file" not in ctx + + def test_the_hook_keeps_a_derive_channel_and_sends_it_back(tmp_path): """Derive keys the server returns land in the session's own `.derive.ids` file and go back as `exclude_derive` on the next write —