feat(write-path): the derive arm — the hint names a duplicate family (no canon) or a canon elsewhere for the shapes being written; exclude_derive channel (#2900, milestone 299 step 2)

shape_ledger.write_time_derive asks the ledger what it knows about each
named (kind, symbol): a derive-grouped family (identical body / same name
in N other files) -> "derive it now, do not add a copy"; a canonical row
at another path -> "canon #N at <path>, reuse". Judged rows at the path and
the canon own file stay silent. Rendered by _derive_line beside the
divergence line; keyed (group id / canon:<id>) on a third per-session dedup
channel in the hook (.derive.ids -> exclude_derive=).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

tests/test_integration_shape_classify.py

@@ -597,6 +597,49 @@ async def test_derive_groups_land_on_rows_and_in_the_summary(seeded):
     assert {r.symbol for r in rows} == {"slug"}                   # 2 files < the name floor
 
 
+@pytest.mark.integration
+async def test_write_time_derive_names_the_family_or_the_canon_for_a_name(seeded):
+    """#2900: against real rows — a name in a dup family → the family (other
+    files, count); a name whose canonical row lives elsewhere → that canon;
+    a judged row at the path, the canon's own file, or an unknown name →
+    silence."""
+    from scribe.services.shape_ledger import apply_derive_groups, write_time_derive
+
+    owner, pid, sid = seeded["owner"], seeded["pid"], seeded["snippet"]
+    defs = _defs(
+        ("v/A.vue", "css", "log-empty", ".log-empty {", ".log-empty { color: red }"),
+        ("v/B.vue", "css", "log-empty", ".log-empty {", ".log-empty { color: red }"),
+        ("v/C.vue", "css", "log-empty", ".log-empty {", ".log-empty { color: red }"),
+        ("src/factory.py", "sym", "factory", "def factory():", "def factory():\n    return 1"),
+    )
+    await sync_repo_shapes(pid, REPO, defs, seen_marker="m1")
+    await classify_shapes(owner, pid, [
+        {"path": "src/factory.py", "symbol": "factory", "status": "canonical", "snippet_id": sid},
+    ])
+    assert await apply_derive_groups(pid) >= 3
+
+    # A 4th copy about to be written → the family, naming the other files.
+    out = await write_time_derive(pid, "v/D.vue", [("css", "log-empty"), ("css", "unknown")])
+    assert len(out) == 1 and out[0]["symbol"] == "log-empty" and out[0]["kind"] == "css"
+    fam = out[0]["family"]
+    assert fam["identical"] is True and fam["label"] == ".log-empty"
+    assert fam["files"] == ["v/A.vue", "v/B.vue", "v/C.vue"] and fam["file_count"] == 3
+    assert out[0]["key"] == fam["group"] and fam["group"].startswith("dup:")
+    # Editing one existing member still names the OTHER members.
+    out = await write_time_derive(pid, "v/A.vue", [("css", "log-empty")])
+    assert out[0]["family"]["files"] == ["v/B.vue", "v/C.vue"] and out[0]["family"]["size"] == 3
+    # The canon's name elsewhere → the canon; in the canon's own file → silence.
+    out = await write_time_derive(pid, "src/other.py", [("sym", "factory")])
+    assert out == [{"symbol": "factory", "kind": "sym", "key": f"canon:{sid}",
+                    "canon": {"snippet_id": sid, "path": "src/factory.py", "label": "factory"}}]
+    assert await write_time_derive(pid, "src/factory.py", [("sym", "factory")]) == []
+    # A judged row at the path is not re-litigated.
+    await classify_shapes(owner, pid, [
+        {"path": "v/B.vue", "symbol": "log-empty", "status": "exempt", "reason": "print sheet"},
+    ])
+    assert await write_time_derive(pid, "v/B.vue", [("css", "log-empty")]) == []
+
+
 @pytest.mark.integration
 async def test_derive_new_names_the_copy_that_joined_a_family_since_the_stamp(seeded):
     """#2899: the first sync seeds one `slug`; a later sync adds an identical

tests/test_write_path_trigger.py

@@ -1243,6 +1243,116 @@ def test_hook_sends_the_enclosing_definition_for_a_body_edit(tmp_path):
     assert seen["shapes"] == ["sym:onTrash"]
 
 
+@pytest.mark.asyncio
+async def test_the_write_time_derive_check_names_a_family_or_a_canon_in_band():
+    """#2900: the ledger's own word on the names being written — a duplicate
+    family with no canon, or a canon recorded elsewhere — rendered at the
+    write even when nothing else does; keyed so the session's exclude
+    channel silences a family already named."""
+    from scribe.services import plugin_context as pc
+    found = [
+        {"symbol": "log-empty", "kind": "css", "key": "dup:483a",
+         "family": {"group": "dup:483a", "label": ".log-empty", "identical": True,
+                    "files": ["a/TaskLogSection.vue", "a/WorkspaceTaskPanel.vue"],
+                    "file_count": 5, "size": 6}},
+        {"symbol": "btn-primary", "kind": "css", "key": "canon:2855",
+         "canon": {"snippet_id": 2855, "path": "frontend/src/assets/components.css",
+                   "label": ".btn-primary"}},
+        {"symbol": "load", "kind": "sym", "key": "name:sym:load",
+         "family": {"group": "name:sym:load", "label": "load", "identical": False,
+                    "files": ["a/X.vue", "a/Y.vue", "a/Z.vue"], "file_count": 3, "size": 4}},
+    ]
+    check = AsyncMock(return_value=found)
+    patches = dict(
+        get_writepath_config=AsyncMock(return_value=_cfg()),
+        semantic_search_notes=AsyncMock(return_value=[]),
+        record_retrieval=MagicMock(), owner_names_for=AsyncMock(return_value={}),
+    )
+    with patch.multiple(pc, **patches), \
+         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", check):
+        out = await pc.build_write_path_hint(
+            1, "frontend/src/components/New.vue", code=REAL_CODE, project_id=24,
+            stamp_shapes=[("css", "log-empty"), ("css", "btn-primary"), ("sym", "load")],
+            exclude_derive=["name:sym:load"],
+        )
+    check.assert_awaited_once_with(24, "frontend/src/components/New.vue",
+                                   [("css", "log-empty"), ("css", "btn-primary"), ("sym", "load")])
+    # The excluded family is gone; the other two render and are keyed.
+    assert [d["key"] for d in out["derive"]] == ["dup:483a", "canon:2855"]
+    assert out["derive_keys"] == ["dup:483a", "canon:2855"]
+    ctx = out["context"]
+    assert "Shape ledger at `frontend/src/components/New.vue`" in ctx
+    assert "`.log-empty` is a duplicate family with no canon — identical body in 5 other file(s): " \
+           "`a/TaskLogSection.vue`, `a/WorkspaceTaskPanel.vue` +3 more; derive it now" in ctx
+    assert "`.btn-primary` is canon — snippet #2855 at `frontend/src/assets/components.css`" in ctx
+    assert "convention-plumbing" in ctx
+    assert "`load`" not in ctx
+    # No project → no check; a failing check never sinks the hint.
+    check.reset_mock()
+    with patch.multiple(pc, **patches), \
+         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", check):
+        out = await pc.build_write_path_hint(1, "x.py", code=REAL_CODE, stamp_shapes=[("sym", "f")])
+    check.assert_not_awaited()
+    assert out["derive"] == [] and out["derive_keys"] == []
+    boom = AsyncMock(side_effect=RuntimeError("ledger down"))
+    with patch.multiple(pc, **patches), \
+         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", boom):
+        out = await pc.build_write_path_hint(1, "x.py", code=REAL_CODE, project_id=24,
+                                             stamp_shapes=[("sym", "f")])
+    assert out["context"] == "" and out["derive"] == []
+
+
+def test_the_hook_keeps_a_derive_channel_and_sends_it_back(tmp_path):
+    """#2900: derive keys the server returns land in the session's own
+    `.derive.ids` file and go back as `exclude_derive` on the next write —
+    a family is named once per session, not at every edit."""
+    import http.server
+    import threading
+    import urllib.parse
+
+    seen: list[dict] = []
+
+    class _Sink(http.server.BaseHTTPRequestHandler):
+        def do_GET(self):
+            seen.append(urllib.parse.parse_qs(urllib.parse.urlparse(self.path).query))
+            self.send_response(200)
+            self.send_header("Content-Type", "application/json")
+            self.end_headers()
+            self.wfile.write(b'{"context":"> family","note_ids":[],"sync_note_ids":[],'
+                             b'"derive_keys":["dup:483a","canon:2855"]}')
+
+        def log_message(self, *a):
+            pass
+
+    server = http.server.HTTPServer(("127.0.0.1", 0), _Sink)
+    threading.Thread(target=server.serve_forever, daemon=True).start()
+    try:
+        env = dict(_hook_runtime_env(), SCRIBE_URL=f"http://127.0.0.1:{server.server_port}",
+                   TMPDIR=str(tmp_path))
+        payload = {"session_id": "s-derive-1", "cwd": str(tmp_path), "tool_name": "Write",
+                   "tool_input": {"file_path": str(tmp_path / "a.css"),
+                                  "content": ".log-empty {\n  color: red;\n}\n"}}
+        for _ in range(2):
+            out = subprocess.run(["bash", str(HOOK)], input=json.dumps(payload),
+                                 capture_output=True, text=True, env=env)
+            assert out.returncode == 0, out.stderr
+    finally:
+        server.shutdown()
+    assert "exclude_derive" not in seen[0]
+    assert seen[1]["exclude_derive"] == ["dup:483a,canon:2855"]
+    state = tmp_path / "scribe-priorart" / "s-derive-1.derive.ids"
+    assert state.read_text().split() == ["dup:483a", "canon:2855", "dup:483a", "canon:2855"]
+
+
 @pytest.mark.asyncio
 async def test_the_write_time_divergence_check_is_named_in_band():
     """#2793: the hook named a shape at a path whose directory a canon