src/scribe/mcp/tools/notes.py

@@ -183,11 +183,14 @@ async def create_note(
             "When Forgejo issues run numbers per workflow rather than per
             repository", not "in six months". Constraints expire when the
             ground moves, not on a schedule.
-        force: Bypass the near-duplicate gate. By default, if a title- or
-            meaning-similar note already exists in the same project, creation is
-            BLOCKED and the existing note's id is returned so you update it
-            instead (no duplicate bloat / no stale RAG copies). Set true only
-            when you're sure this is a genuinely distinct note.
+        force: Bypass the near-duplicate gate. By default, a note with the
+            same title, or one that reads as a copy, in the same project BLOCKS
+            the create and its id is returned so you update it instead. A note
+            that reads CLOSE but not identical does not block: the note is
+            created and the reply carries `overlaps` — open the top one and
+            judge it. Same thing: fold into it and delete the new one. A
+            sibling (the next dev-log, another part of one design): keep both.
+            Set force true only when a blocked note is genuinely distinct.
 
     AN ID EXISTS ONLY ONCE A CREATE RETURNS IT. A body citing a `#N` that has
     not been assigned yet is refused — every session and user draws from one
@@ -203,10 +206,11 @@ async def create_note(
     """
     uid = current_user_id()
     await refuse_guessed_ids(title, body)
+    overlaps: list = []
     if not force:
         dup = await dedup_svc.find_duplicate_note(
             uid, title, body, project_id=project_id or None,
-            is_task=False, note_type="note",
+            is_task=False, note_type="note", overlaps=overlaps,
         )
         if dup is not None:
             return dedup_svc.duplicate_response(dup, "note")
@@ -231,6 +235,7 @@ async def create_note(
     data = note.to_dict()
     await systems_tools.attach_systems(uid, uid, data, note.id, project_id or None)
     await supersession_svc.attach_relations(uid, note.id, data, hint=True)
+    data.update(dedup_svc.note_overlap_response(overlaps, "note"))
     return data
 
 

src/scribe/mcp/tools/tasks.py

@@ -304,10 +304,14 @@ async def create_task(
         arose_from_id: For an issue, the id of the task/feature it arose from;
             for a spike, the record that raised the question — including a
             standing rule whose check just failed. 0 = none.
-        force: Bypass the near-duplicate gate. By default, if a title- or
-            meaning-similar task already exists in the same project, creation is
-            BLOCKED and the existing task's id is returned so you update it
-            instead. Set true only for a genuinely distinct task.
+        force: Bypass the near-duplicate gate. By default, a task with the
+            same title, or one that reads as a copy, in the same project BLOCKS
+            the create and its id is returned so you update it instead. A task
+            that reads CLOSE but not identical does not block: the task is
+            created and the reply carries `overlaps` — open the top one and
+            judge whether it is the same work (fold in, delete the new one) or
+            separate work (keep both). Set force true only when a blocked task
+            is genuinely distinct.
 
     AN ID EXISTS ONLY ONCE A CREATE RETURNS IT. Never write the id you expect
     a record to get: every session and user draws from one sequence, so the
@@ -332,10 +336,11 @@ async def create_task(
             "task with create_task(milestone_id=<that milestone>)."
         )
     await refuse_guessed_ids(title, body)
+    overlaps: list = []
     if not force:
         dup = await dedup_svc.find_duplicate_note(
             uid, title, body, project_id=project_id or None,
-            is_task=True, note_type="note",
+            is_task=True, note_type="note", overlaps=overlaps,
         )
         if dup is not None:
             return dedup_svc.duplicate_response(dup, "task")
@@ -356,6 +361,7 @@ async def create_task(
         await systems_svc.set_record_systems(uid, note.id, system_ids)
     data = note.to_dict()
     await systems_tools.attach_systems(uid, uid, data, note.id, project_id or None)
+    data.update(dedup_svc.note_overlap_response(overlaps, "task"))
     return await placement_svc.attach_placement(uid, data, note)
 
 
@@ -556,13 +562,23 @@ def _batch_items(records: list[dict], *, what: str = "record") -> list[batch_svc
     return items
 
 
-async def _first_duplicate(uid: int, items: list, project_id: int | None) -> dict | None:
-    """The duplicate gate over a whole batch — the first hit blocks all of it."""
+async def _first_duplicate(
+    uid: int, items: list, project_id: int | None,
+    overlaps: dict | None = None,
+) -> dict | None:
+    """The duplicate gate over a whole batch — the first hit blocks all of it.
+
+    `overlaps`, when given, collects each record's near matches below the copy
+    band by its 1-based position (#4306), for the reply once the batch is
+    created."""
     for i, item in enumerate(items, start=1):
+        found: list = []
         dup = await dedup_svc.find_duplicate_note(
             uid, item.title, item.body, project_id=project_id,
-            is_task=item.is_task, note_type="note",
+            is_task=item.is_task, note_type="note", overlaps=found,
         )
+        if found and overlaps is not None:
+            overlaps[i] = found
         if dup is not None:
             payload = dedup_svc.duplicate_response(dup, "task" if item.is_task else "note")
             payload["record"] = i
@@ -616,14 +632,17 @@ async def create_records(
     uid = current_user_id()
     items = _batch_items(records)
     await refuse_guessed_ids(*[t for item in items for t in (item.title, item.body)])
+    overlaps: dict = {}
     if not force:
-        dup = await _first_duplicate(uid, items, project_id or None)
+        dup = await _first_duplicate(uid, items, project_id or None, overlaps)
         if dup is not None:
             return dup
     _ms, notes = await batch_svc.create_batch(
         uid, items, project_id=project_id or None, milestone_id=milestone_id or None,
     )
-    return {"ids": [n.id for n in notes], "records": [n.to_dict() for n in notes]}
+    out = {"ids": [n.id for n in notes], "records": [n.to_dict() for n in notes]}
+    out.update(dedup_svc.batch_overlap_response(overlaps))
+    return out
 
 
 async def start_planning(
@@ -699,14 +718,18 @@ async def start_planning(
         )
         if match is not None:
             return match
+    overlaps: dict = {}
     if items and not force:
-        dup = await _first_duplicate(uid, items, project_id or None)
+        dup = await _first_duplicate(uid, items, project_id or None, overlaps)
         if dup is not None:
             return dup
-    return await planning_svc.start_planning(
+    result = await planning_svc.start_planning(
         user_id=uid, project_id=project_id, title=title,
         body=body or None, steps=items or None,
     )
+    if isinstance(result, dict):
+        result.update(dedup_svc.batch_overlap_response(overlaps))
+    return result
 
 
 async def delete_task(task_id: int) -> dict:

src/scribe/services/dedup.py

@@ -94,6 +94,25 @@ _SNIPPET_SEMANTIC_THRESHOLD = 0.96
 # records that can still be merged by hand.
 _LESSON_SEMANTIC_THRESHOLD = 0.96
 
+# NOTES AND TASKS BLOCK ONLY A COPY, and surface the rest (#4306). Measured
+# 2026-09-22 with find_duplicate_records(note, 0.85): of the 74 note pairs at or
+# above the old 0.90 bar, almost all were DISTINCT siblings — consecutive
+# dev-logs (0.90–0.94), sub-notes of one design (0.90–0.94), research parts
+# (0.90–0.97), lore entries (0.95–0.98). The one clear copy sat at 0.997. A
+# block in that band refused the next dev-log and taught force=true, the same
+# finding #4134 made for rules. So only the copy band blocks; below it, a
+# near match is shown on the create reply for the session to judge.
+_NOTE_COPY_THRESHOLD = 0.98
+# Where a near match starts being worth reading. The measured pair counts
+# climb steeply under 0.87 (36 pairs at 0.87 against 200 capped at 0.85), and
+# the reply lists at most _NOTE_OVERLAP_LIMIT, so this is a cost floor — what
+# decides whether a match matters is the session reading it.
+_NOTE_OVERLAP_FLOOR = 0.87
+_NOTE_OVERLAP_LIMIT = 3
+# The note_types the copy band applies to. A process is prose too, but its
+# gate was not part of the measurement, so it keeps the general bar.
+_COPY_BAND_TYPES = {"note"}
+
 # The gate queries per CHUNK of the candidate (#280) — this caps how many
 # searches one save may cost. Eight chunks ≈ five thousand words of candidate;
 # a duplicate hiding past that is the duplicate report's job to find, not a
@@ -236,9 +255,20 @@ def _semantic_threshold(note_type: str) -> float:
         return _SNIPPET_SEMANTIC_THRESHOLD
     if note_type == LESSON_NOTE_TYPE:
         return _LESSON_SEMANTIC_THRESHOLD
+    if note_type in _COPY_BAND_TYPES:
+        return _NOTE_COPY_THRESHOLD
     return _SEMANTIC_THRESHOLD
 
 
+@dataclass
+class NoteOverlap:
+    """An existing note or task close enough to read before keeping a new one,
+    and not close enough to be called a copy."""
+    id: int
+    title: str
+    similarity: float
+
+
 async def find_duplicate_note(
     user_id: int,
     title: str,
@@ -249,6 +279,7 @@ async def find_duplicate_note(
     code: str = "",
     locations: list[dict] | None = None,
     data: dict | None = None,
+    overlaps: list[NoteOverlap] | None = None,
 ) -> DuplicateMatch | None:
     """Best near-duplicate of (title, body) within the same owner + project +
     kind, or None. Title match first (cheap, exact), then — for snippets — the
@@ -264,6 +295,11 @@ async def find_duplicate_note(
     carries the trigger, which the TITLE no longer does (milestone 427): the
     title check compares names, and the semantic check rebuilds the embedded
     document from `data`.
+
+    `overlaps`, when given, is filled with the near matches below the copy band
+    (#4306) — from the SAME searches, so asking costs nothing extra. Only the
+    kinds in _COPY_BAND_TYPES collect them. The caller creates the record and
+    returns them with `note_overlap_response`.
     """
     norm = " ".join((title or "").split()).lower()
 
@@ -319,6 +355,9 @@ async def find_duplicate_note(
         # under its name and embedded under `name — trigger`, so the query
         # document is built the way the corpus was, from `data`.
         doc_title = embeddings_svc.document_title(title, note_type, data, body)
+        block_at = _semantic_threshold(note_type)
+        collect = overlaps is not None and note_type in _COPY_BAND_TYPES
+        near: dict[int, NoteOverlap] = {}
         for query in embeddings_svc.chunk_document(doc_title, body)[:_GATE_MAX_CHUNKS]:
             # Scope the semantic check the same way as the title check: a record
             # in project P compares only to P; a project-less (orphan) record
@@ -330,7 +369,7 @@ async def find_duplicate_note(
                 user_id, query, project_id=project_id, is_task=is_task,
                 orphan_only=(project_id is None),
                 limit=3,
-                threshold=_semantic_threshold(note_type),
+                threshold=_NOTE_OVERLAP_FLOOR if collect else block_at,
                 # Owner-only, deliberately: this gate BLOCKS a create and tells
                 # the caller to update the match instead. Matching someone
                 # else's record would refuse their write and point them at
@@ -346,12 +385,69 @@ async def find_duplicate_note(
             for score, note in hits:
                 # semantic_search_notes doesn't filter note_type — enforce it
                 # here so a note doesn't shadow a task of the same wording, etc.
-                if note.note_type == note_type:
+                if note.note_type != note_type:
+                    continue
+                if score >= block_at:
                     return DuplicateMatch(note.id, note.title, round(score, 3), "semantic")
+                # Best chunk wins per record: one long note matching in two
+                # sections is one overlap, not two.
+                prior = near.get(note.id)
+                if prior is None or score > prior.similarity:
+                    near[note.id] = NoteOverlap(note.id, note.title, round(score, 3))
+        if collect:
+            overlaps.extend(sorted(
+                near.values(), key=lambda o: o.similarity, reverse=True,
+            )[:_NOTE_OVERLAP_LIMIT])
 
     return None
 
 
+def note_overlap_response(overlaps: list[NoteOverlap], kind: str) -> dict:
+    """The keys a note or task create adds to its reply when an existing record
+    reads closely like the one just written (#4306). Empty when none.
+
+    The judgement is the session's: the embedding cannot tell a restatement
+    from the next dev-log in a series, and a reader can in one look."""
+    if not overlaps:
+        return {}
+    top = overlaps[0]
+    named = "; ".join(f'#{o.id} "{o.title}" ({o.similarity})' for o in overlaps)
+    return {
+        "overlaps": [
+            {"id": o.id, "title": o.title, "similarity": o.similarity}
+            for o in overlaps
+        ],
+        "overlap_note": (
+            f"Created — and it reads closely like: {named}. Open #{top.id} and "
+            f"judge it. If it records the same thing, fold what is new into it "
+            f"(update_{kind}) and delete this {kind}: two copies are found "
+            f"apart and drift apart. If it is a sibling — the next entry in a "
+            f"series, another part of one design — keep both."
+        ),
+    }
+
+
+def batch_overlap_response(per_record: dict[int, list[NoteOverlap]]) -> dict:
+    """`note_overlap_response` for a batch create: each overlap names the
+    1-based record it belongs to, so the reader knows which new id to judge."""
+    rows = [
+        {"record": i, "id": o.id, "title": o.title, "similarity": o.similarity}
+        for i, found in sorted(per_record.items()) for o in found
+    ]
+    if not rows:
+        return {}
+    return {
+        "overlaps": rows,
+        "overlap_note": (
+            "Created — and some records read closely like existing ones (see "
+            "`overlaps`, by record). Open each and judge it: the same thing "
+            "means fold what is new into the existing record and delete the "
+            "new one; a sibling (the next entry in a series, another part of "
+            "one design) means keep both."
+        ),
+    }
+
+
 # --- corpus-wide near-duplicate report (#2088) -------------------------------
 # The gate above PREVENTS a new duplicate; merge_snippets CURES one you point it
 # at. Neither FINDS the duplicates already sitting in the record — someone had to

tests/test_services_dedup.py

@@ -43,8 +43,9 @@ async def test_short_body_skips_semantic_check():
 
 @pytest.mark.asyncio
 async def test_semantic_match_when_body_substantial():
+    # A note blocks only in the copy band (#4306).
     hit = fake_note(id=20, title="Existing", note_type="note")
-    sem = AsyncMock(return_value=[(0.93, hit)])
+    sem = AsyncMock(return_value=[(0.99, hit)])
     with patch("scribe.services.dedup.async_session",
                return_value=session_returning(None)), \
          patch("scribe.services.dedup.embeddings_svc.semantic_search_notes", sem):
@@ -54,7 +55,80 @@ async def test_semantic_match_when_body_substantial():
     assert dup is not None
     assert dup.id == 20
     assert dup.reason == "semantic"
-    assert dup.similarity == 0.93
+    assert dup.similarity == 0.99
+
+
+@pytest.mark.asyncio
+async def test_a_near_note_is_surfaced_not_blocked():
+    """#4306: sibling notes — the next dev-log, another part of one design —
+    measured 0.90–0.98, so a match there is shown for the session to judge
+    instead of refusing the write."""
+    from scribe.services.dedup import _NOTE_OVERLAP_FLOOR
+
+    hit = fake_note(id=20, title="Dev-log day 3", note_type="note")
+    sem = AsyncMock(return_value=[(0.93, hit)])
+    overlaps: list = []
+    with patch("scribe.services.dedup.async_session",
+               return_value=session_returning(None)), \
+         patch("scribe.services.dedup.embeddings_svc.semantic_search_notes", sem):
+        dup = await find_duplicate_note(
+            7, "Dev-log day 4", body="x" * 250, project_id=2, is_task=False,
+            note_type="note", overlaps=overlaps,
+        )
+    assert dup is None
+    assert [(o.id, o.similarity) for o in overlaps] == [(20, 0.93)]
+    # Asked at the overlap floor, so the one search serves both answers.
+    assert sem.await_args.kwargs["threshold"] == _NOTE_OVERLAP_FLOOR
+
+
+@pytest.mark.asyncio
+async def test_a_record_matching_in_two_chunks_is_one_overlap():
+    para = ("A paragraph long enough for the chunker to keep as its own "
+            "section of real content in this test body. ") * 4
+    body = "\n\n".join(f"## Part {i}\n\n{para} (p{i})" for i in range(8))
+    hit = fake_note(id=31, title="Design part one", note_type="note")
+    sem = AsyncMock(return_value=[(0.9, hit)])
+    overlaps: list = []
+    with patch("scribe.services.dedup.async_session",
+               return_value=session_returning(None)), \
+         patch("scribe.services.dedup.embeddings_svc.semantic_search_notes", sem):
+        await find_duplicate_note(
+            7, "Design part two", body=body, project_id=2, note_type="note",
+            overlaps=overlaps,
+        )
+    assert [o.id for o in overlaps] == [31]
+
+
+@pytest.mark.asyncio
+async def test_without_an_overlaps_list_the_gate_asks_at_the_copy_band():
+    """A caller that only wants the block (create_process) does not pay for
+    a wider search it will not read."""
+    from scribe.services.dedup import _NOTE_COPY_THRESHOLD
+
+    sem = AsyncMock(return_value=[])
+    with patch("scribe.services.dedup.async_session",
+               return_value=session_returning(None)), \
+         patch("scribe.services.dedup.embeddings_svc.semantic_search_notes", sem):
+        await find_duplicate_note(7, "T", body="x" * 250, note_type="note")
+    assert sem.await_args.kwargs["threshold"] == _NOTE_COPY_THRESHOLD
+
+
+def test_the_overlap_reply_leaves_the_judgement_to_the_reader():
+    from scribe.services.dedup import NoteOverlap, note_overlap_response
+
+    assert note_overlap_response([], "note") == {}
+    out = note_overlap_response([NoteOverlap(9, "Dev-log day 3", 0.93)], "task")
+    assert out["overlaps"] == [{"id": 9, "title": "Dev-log day 3", "similarity": 0.93}]
+    assert "update_task" in out["overlap_note"]
+    assert "keep both" in out["overlap_note"]
+
+
+def test_a_batch_overlap_names_its_record():
+    from scribe.services.dedup import NoteOverlap, batch_overlap_response
+
+    assert batch_overlap_response({}) == {}
+    out = batch_overlap_response({2: [NoteOverlap(9, "X", 0.9)]})
+    assert out["overlaps"] == [{"record": 2, "id": 9, "title": "X", "similarity": 0.9}]
 
 
 @pytest.mark.asyncio
@@ -73,7 +147,7 @@ async def test_gate_catches_a_duplicate_hiding_in_a_later_chunk():
 
     hit = fake_note(id=30, title="The existing decision", note_type="note")
     # Every chunk misses except the LAST one the gate will ask about.
-    sem = AsyncMock(side_effect=[[] for _ in range(n_chunks - 1)] + [[(0.94, hit)]])
+    sem = AsyncMock(side_effect=[[] for _ in range(n_chunks - 1)] + [[(0.99, hit)]])
     with patch("scribe.services.dedup.async_session",
                return_value=session_returning(None)), \
          patch("scribe.services.dedup.embeddings_svc.semantic_search_notes", sem):