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 <noreply@anthropic.com>

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

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

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

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 —