Merge pull request 'Milestone 415: an existing plan is found before a new one is made' (#159) from dev into main

alembic/versions/0102_milestone_embeddings.py

@@ -0,0 +1,59 @@
+"""milestone_embeddings — a plan becomes findable by meaning (milestone 415)
+
+Revision ID: 0102
+Revises: 0101
+Create Date: 2026-09-15
+
+`search` covered notes, tasks and rules, and a milestone — the record a plan
+lives in — could not be found at all. So "is there already a plan for this?"
+had no tool, and a project whose roadmap was written as milestones had every
+later plan opened as a new milestone beside the one that already described it.
+
+The sibling of rule_embeddings (0089), for the reasons its model docstring and
+note 3163 give. The vectors are DERIVED: nothing is backfilled here, the startup
+backfill writes them.
+"""
+import sqlalchemy as sa
+from alembic import op
+
+revision = "0102"
+down_revision = "0101"
+branch_labels = None
+depends_on = None
+
+# Matches note_embeddings and rule_embeddings — bge-small-en-v1.5, 384-dim.
+_EMBEDDING_DIM = 384
+
+
+def upgrade() -> None:
+    op.create_table(
+        "milestone_embeddings",
+        sa.Column(
+            "milestone_id", sa.Integer(),
+            sa.ForeignKey("milestones.id", ondelete="CASCADE"), primary_key=True,
+        ),
+        sa.Column("chunk_index", sa.Integer(), primary_key=True),
+        sa.Column("chunk_text", sa.Text(), nullable=False),
+        sa.Column("chunker_version", sa.Integer(), nullable=False),
+        sa.Column(
+            "updated_at", sa.DateTime(timezone=True), nullable=False,
+            server_default=sa.text("now()"),
+        ),
+    )
+    # Raw DDL for the vector column, as 0067 and 0089 do: the type comes from
+    # the pgvector extension, not SQLAlchemy's type system.
+    op.execute(
+        f"ALTER TABLE milestone_embeddings ADD COLUMN embedding vector({_EMBEDDING_DIM}) NOT NULL"
+    )
+    op.execute(
+        """
+        CREATE INDEX ix_milestone_embeddings_embedding_hnsw
+            ON milestone_embeddings
+         USING hnsw (embedding vector_cosine_ops)
+        """
+    )
+
+
+def downgrade() -> None:
+    op.execute("DROP INDEX IF EXISTS ix_milestone_embeddings_embedding_hnsw")
+    op.drop_table("milestone_embeddings")

docs/features.md

@@ -45,6 +45,10 @@ Tasks carry status (`todo` → `in_progress` → `done`/`cancelled`), priority
 - **Milestones** — Ordered stages within a project. A milestone is also the home of a
   **plan** — its body holds the design (Goal/Approach/Verification) and its child
   tasks are the steps. Completion percentage is shown on the project page.
+  Milestones are searchable by meaning (`search(content_type="milestone")`), and an
+  agent starting a plan (`start_planning`, `create_milestone`) is handed the active
+  milestone that already has its title or reads as the same plan, so steps are added
+  there rather than to a parallel plan. The match threshold is in Settings.
 - **Kanban view** — `/projects/:id` groups tasks by milestone in a column layout with
   status-advance buttons on the cards.
 

frontend/src/views/ProjectView.vue

@@ -179,6 +179,25 @@ const milestoneGroups = computed((): MilestoneGroup[] => {
   return groups;
 });
 
+// The "No Milestone" group has no id, and collapsing it uses the same Set as
+// the milestones: 0 stands for it, since no milestone id is ever 0.
+const UNASSIGNED_KEY = 0;
+function groupKey(group: MilestoneGroup): number {
+  return group.milestone?.id ?? UNASSIGNED_KEY;
+}
+
+// A group's Done column starts COLLAPSED, to its header and count. Done work is
+// the part of a board nobody is reading, and listed in full it pushed the open
+// columns of every group below it off the screen (operator, 2026-09-15). Not
+// persisted, matching the milestone collapse beside it: each load starts from
+// the same defaults.
+const expandedDone = ref<Set<number>>(new Set());
+function toggleDone(group: MilestoneGroup) {
+  const key = groupKey(group);
+  if (expandedDone.value.has(key)) expandedDone.value.delete(key);
+  else expandedDone.value.add(key);
+}
+
 function toggleMilestoneCollapse(id: number) {
   if (collapsedMilestones.value.has(id)) {
     collapsedMilestones.value.delete(id);
@@ -386,6 +405,16 @@ async function loadTasks() {
       all.push(...next.notes);
     }
     tasks.value = all;
+    // The No Milestone group follows the milestone rule: once, on first load,
+    // it starts collapsed when everything in it is finished.
+    if (!autoCollapsedOnce.value.has(UNASSIGNED_KEY)) {
+      autoCollapsedOnce.value.add(UNASSIGNED_KEY);
+      const assigned = new Set(milestones.value.map((m) => m.id));
+      const loose = all.filter((t) => !t.milestone_id || !assigned.has(t.milestone_id));
+      if (loose.length && loose.every((t) => t.status === "done" || t.status === "cancelled")) {
+        collapsedMilestones.value.add(UNASSIGNED_KEY);
+      }
+    }
   } catch {
     // Say so. This used to swallow the error and leave an empty board, which is
     // indistinguishable from a project with no tasks — the same "hidden with no
@@ -950,11 +979,12 @@ async function confirmDelete() {
               <div v-for="group in milestoneGroups" :key="group.milestone?.id ?? 'unassigned'" class="milestone-group">
                 <div
                   class="milestone-header"
-                  :class="{ clickable: !!group.milestone && renamingMilestoneId !== group.milestone?.id }"
-                  @click="group.milestone && renamingMilestoneId !== group.milestone.id && toggleMilestoneCollapse(group.milestone.id)"
+                  :class="{ clickable: renamingMilestoneId !== group.milestone?.id }"
+                  :aria-expanded="!collapsedMilestones.has(groupKey(group))"
+                  @click="renamingMilestoneId !== group.milestone?.id && toggleMilestoneCollapse(groupKey(group))"
                 >
-                  <span class="ms-chevron" v-if="group.milestone">
-                    <ChevronRight v-if="collapsedMilestones.has(group.milestone.id)" :size="16" />
+                  <span class="ms-chevron">
+                    <ChevronRight v-if="collapsedMilestones.has(groupKey(group))" :size="16" />
                     <ChevronDown v-else :size="16" />
                   </span>
                   <template v-if="group.milestone && renamingMilestoneId === group.milestone.id">
@@ -1023,7 +1053,7 @@ async function confirmDelete() {
                   </template>
                 </div>
 
-                <div v-if="!group.milestone || !collapsedMilestones.has(group.milestone.id)" class="kanban">
+                <div v-if="!collapsedMilestones.has(groupKey(group))" class="kanban">
                   <!-- Todo column -->
                   <div class="kanban-col col-todo">
                     <div class="kanban-col-header">
@@ -1094,12 +1124,19 @@ async function confirmDelete() {
 
                   <!-- Done column -->
                   <div class="kanban-col col-done">
-                    <div class="kanban-col-header">
+                    <button
+                      type="button"
+                      class="kanban-col-header col-toggle"
+                      :aria-expanded="expandedDone.has(groupKey(group))"
+                      @click="toggleDone(group)"
+                    >
                       <span class="col-status-dot dot-done"></span>
                       <span class="col-label">Done</span>
                       <span class="col-count">{{ group.tasks.filter(t => t.status === 'done').length }}</span>
-                    </div>
-                    <div class="kanban-cards">
+                      <ChevronDown v-if="expandedDone.has(groupKey(group))" :size="14" />
+                      <ChevronRight v-else :size="14" />
+                    </button>
+                    <div v-if="expandedDone.has(groupKey(group))" class="kanban-cards">
                       <router-link
                         v-for="task in group.tasks.filter(t => t.status === 'done')"
                         :key="task.id" :to="`/tasks/${task.id}`"
@@ -1679,6 +1716,18 @@ async function confirmDelete() {
 }
 .col-status-dot { width: 7px; height: 7px; border-radius: 50%; flex-shrink: 0; }
 .col-label { flex: 1; }
+/* The Done column's header is a button that folds its cards. It keeps the
+   header's look — the reset is only what a <button> brings with it. */
+.col-toggle {
+  width: 100%;
+  background: none;
+  border: none;
+  padding: 0;
+  font-family: inherit;
+  cursor: pointer;
+  text-align: left;
+}
+.col-toggle:focus-visible { outline: none; box-shadow: var(--fs-focus-ring); border-radius: var(--fs-radius-sm); }
 .col-count {
   background: var(--fs-surface-raised);
   border: 1px solid var(--fs-border-color);

frontend/src/views/SettingsView.vue

@@ -104,6 +104,8 @@ const kbPromptRuleThreshold = ref("0.72");
 const kbDupThresholdSnippet = ref("0.82");
 const kbDupThresholdNote = ref("0.93");
 const kbDupThresholdTask = ref("0.93");
+// PLAN_MATCH_DEFAULT_THRESHOLD in services/dedup.py.
+const kbPlanMatchThreshold = ref("0.90");
 const savingKbInject = ref(false);
 const kbInjectSaved = ref(false);
 
@@ -155,6 +157,9 @@ async function saveKbInject() {
   const dupSnip = Math.min(1, Math.max(0, Number(kbDupThresholdSnippet.value) || 0.82));
   const dupNote = Math.min(1, Math.max(0, Number(kbDupThresholdNote.value) || 0.93));
   const dupTask = Math.min(1, Math.max(0, Number(kbDupThresholdTask.value) || 0.93));
+  // Same `|| default` guard: a floor of 0 would hand back an existing plan
+  // for every new one, and no plan could be started without force.
+  const planT = Math.min(1, Math.max(0, Number(kbPlanMatchThreshold.value) || 0.9));
   // Same `|| default` reasoning: falling back to 0 would surface every
   // snippet in the corpus on every edit, which is the failure this knob fixes.
   const wpT = Math.min(1, Math.max(0, Number(kbWritePathThreshold.value) || 0.68));
@@ -171,6 +176,7 @@ async function saveKbInject() {
   kbDupThresholdSnippet.value = String(dupSnip);
   kbDupThresholdNote.value = String(dupNote);
   kbDupThresholdTask.value = String(dupTask);
+  kbPlanMatchThreshold.value = String(planT);
   kbWritePathThreshold.value = String(wpT);
   kbRuleHintThreshold.value = String(rhT);
   kbToolRuleThreshold.value = String(trT);
@@ -199,6 +205,7 @@ async function saveKbInject() {
       kb_duplicate_threshold_snippet: String(dupSnip),
       kb_duplicate_threshold_note: String(dupNote),
       kb_duplicate_threshold_task: String(dupTask),
+      kb_plan_match_threshold: String(planT),
     });
     kbInjectSaved.value = true;
     setTimeout(() => (kbInjectSaved.value = false), 2000);
@@ -662,6 +669,9 @@ onMounted(async () => {
   if (allSettings.kb_duplicate_threshold_task !== undefined) {
     kbDupThresholdTask.value = allSettings.kb_duplicate_threshold_task;
   }
+  if (allSettings.kb_plan_match_threshold !== undefined) {
+    kbPlanMatchThreshold.value = allSettings.kb_plan_match_threshold;
+  }
   if (allSettings.notify_task_reminders !== undefined) {
     notifyTaskReminders.value = allSettings.notify_task_reminders !== "false";
   }
@@ -1623,6 +1633,27 @@ async function deleteUser(userId: number) {
             stays strict to keep the report pointed at work opened twice.
           </p>
         </div>
+
+        <div class="field">
+          <label for="kb-plan-match-threshold">Existing-plan match threshold</label>
+          <input
+            id="kb-plan-match-threshold"
+            v-model="kbPlanMatchThreshold"
+            type="number"
+            min="0"
+            max="1"
+            step="0.01"
+            class="fs-input input"
+            style="max-width: 8rem"
+          />
+          <p class="field-hint">
+            How alike a new plan must be to an active milestone in the same project
+            before an agent is handed that milestone instead of creating a second
+            one. A plan with the same title always matches. Lower it if sessions
+            still open parallel plans for work that already has one; raise it if
+            they are sent to plans that are only related.
+          </p>
+        </div>
         <div class="actions">
           <button class="btn-primary" @click="saveKbInject" :disabled="savingKbInject">
             {{ savingKbInject ? 'Saving…' : 'Save' }}

plugin/.claude-plugin/plugin.json

@@ -1,7 +1,7 @@
 {
   "name": "scribe",
   "description": "Scribe for Claude Code: connects the scribe MCP server, adds the hooks that deliver live project state and relevant records at the right moment, ships the shared client-neutral Scribe skills (using-scribe, writing-plans, reporting-back, systematic-debugging, verification, brainstorming, reusing-code, shape-accounting), and syncs your saved Scribe Processes as skills (/scribe:sync).",
-  "version": "2026.09.15.1626",
+  "version": "2026.09.15.1744",
   "author": {
     "name": "Bryan Van Deusen"
   },

plugin/skills/using-scribe/SKILL.md

@@ -111,6 +111,12 @@ Two constraints on *how* that's achieved:
    is just a task — don't wrap it in a milestone. Either way, do not write
    plans/specs to local `.md` files. See the **writing-plans** skill.
 
+   **Find the plan before you make one.** A project's existing milestones are
+   often its roadmap, and a milestone with no steps yet is still open work.
+   `search(content_type="milestone", project_id=...)` finds one by purpose. When
+   an active milestone covers the work, add steps to it rather than opening
+   another, and give any task you record for that work its `milestone_id`.
+
 5. **Keep state honest.** Set a task `in_progress` when you start it, `done` the
    moment it's complete; log progress as you go. Always log when you
    **complete** a task and when you **hit or discover a problem**, so a change

plugin/skills/writing-plans/SKILL.md

@@ -26,6 +26,34 @@ a complete, honest record of work that didn't need a plan.
 Some projects are milestone-shaped and some are a flat task list. Read the
 project you are in rather than imposing a shape on it.
 
+## Then look for the plan that already exists
+
+Before you start a plan, find out whether the project already has one for this
+work. A roadmap is often written ahead of the work as milestones with a goal
+and no steps yet, and those are exactly the plans a session misses: they have
+nothing in them to show up as open tasks.
+
+- `enter_project` lists the recent milestones and, under `unplanned_milestones`,
+  the active ones with no steps.
+- `search(content_type="milestone", project_id=...)` finds a plan by what it is
+  for. `list_milestones` reads the whole roadmap.
+
+**When an active milestone already covers the work, the plan is that milestone.**
+Read it with `get_milestone`, add your steps with
+`create_records(milestone_id=<its id>, records=[…])`, and revise its body with
+`update_milestone` if what you know now changes the design. Opening a new
+milestone beside it splits one piece of work across two plans, and neither
+shows the whole. A second milestone is right only when the work is a separate
+arc: a different goal that would still make sense if the first plan were done.
+
+The same holds for single tasks. Work that belongs to an existing plan is
+created with that milestone's `milestone_id`, not left loose beside it.
+
+`start_planning` and `create_milestone` check too. When an active milestone in
+the project has the same title or reads as the same plan, they return it as
+`existing_milestone` and create nothing. Add to that plan, or pass `force=true`
+only when you have read it and this really is separate work.
+
 ## When it does: start the plan in Scribe, not a file
 
 Call **`start_planning(project_id, title)`** before designing or implementing —
@@ -59,7 +87,7 @@ record, not a file on disk. (The old `kind=plan` task is retired; `start_plannin
 no longer creates one.)
 
 Before designing from scratch, **recall**: `search` Scribe for a related prior
-plan or decision. Often the thinking (or half of it) already exists.
+decision or note. Often the thinking (or half of it) already exists.
 
 ## What a good plan contains
 

src/scribe/app.py

@@ -161,7 +161,9 @@ def create_app() -> Quart:
         import asyncio
 
         from scribe.services.auth import start_auth_token_retention_loop
-        from scribe.services.embeddings import backfill_note_embeddings, backfill_rule_embeddings
+        from scribe.services.embeddings import (
+            backfill_milestone_embeddings, backfill_note_embeddings, backfill_rule_embeddings,
+        )
         from scribe.services.logging import start_log_retention_loop
         from scribe.services.notifications import start_notification_loop
 
@@ -182,6 +184,12 @@ def create_app() -> Quart:
                 await backfill_rule_embeddings()
             except Exception:
                 logger.warning("Rule embedding backfill failed", exc_info=True)
+            # Milestones got vectors in milestone 415, so a plan written before
+            # it is findable only after this pass.
+            try:
+                await backfill_milestone_embeddings()
+            except Exception:
+                logger.warning("Milestone embedding backfill failed", exc_info=True)
             # Snippets written before migration 0070 have no `notes.data` mirror,
             # and the location reverse lookup queries that column — an unfilled
             # row would read as "no snippet here" rather than as a gap. Separate

src/scribe/mcp/server.py

@@ -49,8 +49,9 @@ client reads Agent Skills) and in each tool's description. The index:
 - RECALL: search before acting, scoped with the active project_id.
 - RECORD: create_task; a fix is kind="issue". add_task_log as you go; status
   in_progress on start, done on finish. Tag system_ids as you write.
-- PLAN work with an arc: start_planning(steps=[...]). The plan is a milestone
-  and each step a task.
+- PLAN work with an arc: find the existing plan first
+  (search(content_type="milestone")) and add steps to it; else
+  start_planning(steps=[...]). The plan is a milestone and each step a task.
 - IDS exist only once a create returns them. Records that cite each other go
   through create_records, writing {{ref:N}} for the Nth record.
 - REUSE: search snippets before building; create_snippet what you build.

src/scribe/mcp/tools/milestones.py

@@ -12,6 +12,7 @@ Sentinels:
 from __future__ import annotations
 
 from scribe.mcp._context import current_user_id
+from scribe.services import dedup as dedup_svc
 from scribe.services import milestones as milestones_svc
 from scribe.services import notes as notes_svc
 from scribe.services import rulebooks as rulebooks_svc
@@ -71,6 +72,7 @@ async def create_milestone(
     description: str = "",
     body: str = "",
     status: str = "active",
+    force: bool = False,
 ) -> dict:
     """Create a milestone within a Scribe project.
 
@@ -85,9 +87,21 @@ async def create_milestone(
         description: Optional one-line summary of what this milestone covers.
         body: Optional plan/design (markdown) — the milestone's full plan text.
         status: active (default) or done.
+        force: Bypass the plan gate. By default an active milestone that
+            already has this title, or reads as the same plan, is returned
+            (`existing_milestone`) and nothing is created.
     """
     uid = current_user_id()
     await refuse_guessed_ids(title, description, body)
+    # A done milestone is a record of past work, not a competing plan, so
+    # only an active create is gated — the same line find_matching_plan draws.
+    if status == "active" and not force:
+        match = await dedup_svc.plan_gate(
+            uid, project_id, title,
+            dedup_svc.plan_candidate_text(description=description, body=body),
+        )
+        if match is not None:
+            return match
     milestone = await milestones_svc.create_milestone(
         uid,
         project_id=project_id,

src/scribe/mcp/tools/projects.py

@@ -48,6 +48,7 @@ async def list_projects() -> dict:
 # characters, past what a client accepts as a tool result (#4045).
 _HANDSHAKE_MILESTONES = 5
 _HANDSHAKE_OPEN_TASKS = 10
+_HANDSHAKE_UNPLANNED = 10
 
 
 async def enter_project(project_id: int) -> dict:
@@ -67,7 +68,8 @@ async def enter_project(project_id: int) -> dict:
 
     Returns a dict with keys: project, milestone_summary, open_tasks, systems,
     design_system, project_rules, pattern_coverage —
-    plus milestone_summary_omitted, inception and systems_bootstrap, each
+    plus unplanned_milestones, milestone_summary_omitted,
+    unplanned_milestones_omitted, inception and systems_bootstrap, each
     present only when it applies (see below).
 
     `project` is id, title, status and the full goal. get_project has the
@@ -79,6 +81,15 @@ async def enter_project(project_id: int) -> dict:
     get_milestone(id) reads a plan and its steps. `milestone_summary_omitted`
     says how many others exist; list_milestones lists them all.
 
+    `unplanned_milestones` is the active milestones that have NO steps yet,
+    in roadmap order (up to 10; `unplanned_milestones_omitted` counts the
+    rest) — id, title and description. They are open work: a plan somebody
+    wrote down and nobody has broken into steps. A milestone with no steps is
+    never touched, so the recent list above can never show one. Before
+    starting a new milestone for work, check whether one of these already
+    describes it; if so, add steps to it with create_records(milestone_id=…)
+    rather than opening a second plan for the same thing.
+
     `open_tasks` is the 10 most recently touched todo / in-progress tasks,
     with or without a milestone. A work-log counts as touching its task. Each
     names its milestone. list_tasks has the rest.
@@ -156,6 +167,11 @@ async def enter_project(project_id: int) -> dict:
         milestone_rows, limit=_HANDSHAKE_MILESTONES,
     )
     milestone_titles = {m["id"]: m.get("title") for m in milestone_rows}
+    unplanned, unplanned_omitted = milestones_svc.unplanned_milestones(
+        milestone_rows,
+        exclude_ids={m["id"] for m in milestone_summary},
+        limit=_HANDSHAKE_UNPLANNED,
+    )
     open_tasks, _ = await notes_svc.list_notes(
         uid, is_task=True, project_id=project_id,
         status=["todo", "in_progress"], sort="touched", limit=_HANDSHAKE_OPEN_TASKS,
@@ -256,6 +272,13 @@ async def enter_project(project_id: int) -> dict:
             f"{omitted} other milestone(s) not listed. "
             f"list_milestones({project_id}) lists every milestone."
         )
+    if unplanned:
+        out["unplanned_milestones"] = unplanned
+    if unplanned_omitted:
+        out["unplanned_milestones_omitted"] = (
+            f"{unplanned_omitted} more active milestone(s) with no steps. "
+            f"list_milestones({project_id}) lists every milestone."
+        )
     if systems_bootstrap:
         out["systems_bootstrap"] = systems_bootstrap
     if inception_ask:

src/scribe/mcp/tools/search.py

@@ -12,7 +12,8 @@ import time
 from scribe.mcp._context import current_user_id
 from scribe.services.access import owner_names_for
 from scribe.services.embeddings import (
-    DEFAULT_SIMILARITY_THRESHOLD, semantic_search_notes, semantic_search_rules,
+    DEFAULT_SIMILARITY_THRESHOLD, semantic_search_milestones, semantic_search_notes,
+    semantic_search_rules,
 )
 from scribe.services import rulebooks as rulebooks_svc
 from scribe.services.retrieval_telemetry import record_retrieval, retrieval_summary
@@ -67,6 +68,40 @@ async def _search_rules(uid: int, q: str, limit: int, project_id: int) -> dict:
     }
 
 
+async def _search_milestones(uid: int, q: str, limit: int, project_id: int) -> dict:
+    """Milestones by meaning — "is there already a plan for this?" (milestone 415).
+
+    Its own result shape, like rules: a milestone is a plan with progress, not
+    a note with a body. The plan itself is left out — get_milestone reads it —
+    because a search hit is for recognising a plan, and bodies run long.
+    Not part of content_type="all", whose results are note-shaped.
+    """
+    raw = await semantic_search_milestones(uid, q, project_id=project_id or None, limit=limit)
+    progress: dict[int, dict] = {}
+    if raw:
+        from scribe.services import milestones as milestones_svc
+
+        for pid in {m.project_id for _s, m in raw}:
+            for row in await milestones_svc.get_project_milestone_summary(uid, pid):
+                progress[row["id"]] = row
+    return {
+        "results": [
+            {
+                "id": m.id,
+                "title": m.title,
+                "description": m.description or "",
+                "status": m.status,
+                "project_id": m.project_id,
+                "total": progress.get(m.id, {}).get("total", 0),
+                "completed": progress.get(m.id, {}).get("completed", 0),
+                "similarity": float(score),
+            }
+            for score, m in raw
+        ],
+        "total": len(raw),
+    }
+
+
 async def search(
     q: str,
     content_type: str = "all",
@@ -93,7 +128,12 @@ async def search(
             tagging?". A hit carries the rule's `why` and `how_to_apply`,
             which the session-start payload does not. With a project_id,
             rules come back as the global rules plus that project's own;
-            with 0, every rule in the rulebook.
+            with 0, every rule in the rulebook. Or 'milestone' (PLANS):
+            reach for it before start_planning to ask whether a plan for
+            this work already exists — a match is where new steps go
+            (create_records(milestone_id=…)), not a reason to open a second
+            milestone. Hits carry title, description, status and progress;
+            get_milestone reads the plan. Not included in 'all'.
         limit: maximum number of results (1-50).
         project_id: Scope results to one project. PASS THE ACTIVE PROJECT'S ID
             whenever a project is in scope (the one you entered with
@@ -118,6 +158,8 @@ async def search(
     limit = max(1, min(limit, 50))
     if content_type == "rule":
         return await _search_rules(uid, q, limit, project_id)
+    if content_type == "milestone":
+        return await _search_milestones(uid, q, limit, project_id)
     is_task = {"note": False, "task": True}.get(content_type)  # None => any
     t0 = time.perf_counter()
     report: dict = {}

src/scribe/mcp/tools/tasks.py

@@ -152,6 +152,9 @@ async def create_task(
         priority: One of: low, medium, high, or 'none'. Omit (empty string) to leave unset.
         project_id: Associate with a project (0 = no project).
         milestone_id: Place within a project milestone (0 = no milestone).
+            When the work belongs to an active plan — a milestone in
+            enter_project's lists or found by search(content_type=
+            "milestone") — pass its id, so the plan shows all of its work.
         parent_id: Make this a sub-task of another task (0 = top-level).
         tags: List of plain-string tags without # prefix.
         kind: 'work' (default), 'issue', or 'spike'.
@@ -462,6 +465,8 @@ async def create_records(
         project_id: The project every record belongs to (0 = none, or taken
             from milestone_id).
         milestone_id: File every record under this existing milestone (0 = none).
+            This is how steps are added to a plan that already exists,
+            including one start_planning handed back as `existing_milestone`.
         force: Bypass the near-duplicate gate for the whole batch. By default
             the first record that near-duplicates an existing one BLOCKS the
             batch, and its existing id comes back so you can update it instead.
@@ -531,17 +536,33 @@ async def start_planning(
         steps: The plan's step-tasks, in order — each an object with `title`
             (required) and optionally `body`, `status`, `priority`, `kind`
             ('work' | 'issue' | 'spike'), `tags`, `system_ids`.
-        force: Bypass the near-duplicate gate on the steps. By default a step
-            that near-duplicates an existing task blocks the whole plan, and
-            nothing — milestone included — is created.
+        force: Bypass both duplicate gates. By default nothing is created —
+            milestone included — when an ACTIVE plan in the project already
+            has this title or reads as the same plan, or when a step
+            near-duplicates an existing task. Pass it once you have read the
+            match and know this is separate work.
 
     Returns the milestone, the project's applicable rules and brief context,
     plus `steps` (the created tasks, in order) when steps were given — OR a
-    duplicate payload naming the `record` that matched, with nothing created.
+    duplicate payload with nothing created: `existing_milestone` when a plan
+    already covers this (add your steps to it with create_records(
+    milestone_id=...)), or `record` naming the step that matched a task.
     """
     uid = current_user_id()
     items = _batch_items(steps or [], what="step")
     await refuse_guessed_ids(body, *[t for item in items for t in (item.title, item.body)])
+    if not force:
+        # The plan before its steps: when a plan already covers this, the
+        # answer is to add these steps THERE, and a step-level match would
+        # only name one symptom of that.
+        match = await dedup_svc.plan_gate(
+            uid, project_id, title,
+            dedup_svc.plan_candidate_text(
+                body=body, steps=[(i.title, i.body) for i in items],
+            ),
+        )
+        if match is not None:
+            return match
     if items and not force:
         dup = await _first_duplicate(uid, items, project_id or None)
         if dup is not None:

src/scribe/models/__init__.py

@@ -25,7 +25,7 @@ from scribe.models.user import User  # noqa: E402, F401
 from scribe.models.app_log import AppLog  # noqa: E402, F401
 from scribe.models.password_reset import PasswordResetToken  # noqa: E402, F401
 from scribe.models.invitation import InvitationToken  # noqa: E402, F401
-from scribe.models.embedding import NoteEmbedding, RuleEmbedding  # noqa: E402, F401
+from scribe.models.embedding import MilestoneEmbedding, NoteEmbedding, RuleEmbedding  # noqa: E402, F401
 from scribe.models.retrieval_log import RetrievalLog  # noqa: E402, F401
 from scribe.models.note_usage import NoteUsageEvent  # noqa: E402, F401
 from scribe.models.rule_usage import RuleUsageEvent  # noqa: E402, F401

src/scribe/models/embedding.py

@@ -95,3 +95,36 @@ class RuleEmbedding(Base):
         DateTime(timezone=True),
         default=lambda: datetime.now(timezone.utc),
     )
+
+
+class MilestoneEmbedding(Base):
+    """One embedding vector per CHUNK of a milestone (milestone 415).
+
+    The third sibling, for note 3163's reason: the row could be shared, the
+    search cannot. A milestone is scoped by its project, has no share of its
+    own, and is searched to answer one question — "is there already a plan for
+    this?" — which no note or rule search can answer, because a plan is not a
+    note. Before this, a roadmap written as milestones was invisible to recall,
+    and every later plan was opened as a new milestone beside the one that
+    already described it.
+
+    The document is the title, the one-line description and the plan body, the
+    parts a reader uses to recognise a plan. Derived data: the startup backfill
+    regenerates it, which is also how a chunker-version bump is handled.
+    """
+
+    __tablename__ = "milestone_embeddings"
+
+    milestone_id: Mapped[int] = mapped_column(
+        Integer,
+        ForeignKey("milestones.id", ondelete="CASCADE"),
+        primary_key=True,
+    )
+    chunk_index: Mapped[int] = mapped_column(Integer, primary_key=True)
+    embedding: Mapped[list] = mapped_column(Vector(EMBEDDING_DIM), nullable=False)
+    chunk_text: Mapped[str] = mapped_column(Text, nullable=False)
+    chunker_version: Mapped[int] = mapped_column(Integer, nullable=False)
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        default=lambda: datetime.now(timezone.utc),
+    )

src/scribe/services/backup.py

@@ -115,7 +115,8 @@ _BACKED_UP = [
 # like coverage while naming nothing the schema could confirm.
 _NOT_INCLUDED = [
     "groups", "group_memberships", "project_shares", "note_shares",
-    "api_keys", "note_embeddings", "rule_embeddings", "app_logs", "notifications",
+    "api_keys", "note_embeddings", "rule_embeddings", "milestone_embeddings",
+    "app_logs", "notifications",
     "invitation_tokens", "password_reset_tokens", "user_profiles",
     "retrieval_logs",
     # Sensitive credentials, same reasoning as api_keys: a backup that carries

src/scribe/services/dedup.py

@@ -33,6 +33,7 @@ from scribe.models.embedding import NoteEmbedding
 from scribe.models.note import Note
 from scribe.models.rulebook import Rule
 from scribe.models.base import iso
+from scribe.services.access import can_read_project
 from scribe.services import embeddings as embeddings_svc
 # Imported rather than redeclared: no service imports this module (the create
 # gate is called from the routes/tools layer), so there is no cycle to dodge,
@@ -720,3 +721,176 @@ async def find_duplicate_rule(
     except Exception:
         logger.debug("dedup rule title check skipped — query failed", exc_info=True)
     return None
+
+
+# --- the plan gate (milestone 415) -------------------------------------------
+# A session asked "what work is open?" that cannot see an existing plan makes a
+# second one: a new milestone beside the one that already covers the work, or
+# loose tasks beside it. Each copy then collects its own steps, and neither
+# shows the whole. This gate asks the question before start_planning (or
+# create_milestone) writes: is there an ACTIVE plan in this project for this?
+#
+# Active only: a done milestone is history, and planning the next round of
+# the same area is legitimate work rather than a copy of it.
+#
+# Project-scoped, not owner-scoped like the note gate. The note gate refuses to
+# point at someone else's record because they may not be able to edit it; a
+# plan is different. Creating one needs write on the project, and write on the
+# project is exactly what adding steps to its existing plan needs, so a caller
+# who reaches this gate can act on whatever it returns.
+#
+# Its own threshold, as a setting (rule 25). The note gate's 0.90 is where it
+# starts, because it asks the same question ("the same thing, reworded") with
+# the same embedder. Plan documents are shaped differently, though: a milestone
+# is embedded as title, description and plan (embeddings.milestone_document),
+# while the candidate usually has no description and carries its steps
+# instead. That difference has not been measured, and a gate that blocks on
+# noise teaches sessions to pass force=true every time, which is worse than no
+# gate. So the default is conservative and the operator can lower it.
+PLAN_MATCH_THRESHOLD_KEY = "kb_plan_match_threshold"
+PLAN_MATCH_DEFAULT_THRESHOLD = 0.90
+
+
+async def get_plan_match_threshold(user_id: int) -> float:
+    """The user's plan-gate similarity floor, clamped to [0, 1]."""
+    from scribe.services.settings import get_setting
+
+    try:
+        value = float(await get_setting(
+            user_id, PLAN_MATCH_THRESHOLD_KEY, str(PLAN_MATCH_DEFAULT_THRESHOLD)
+        ))
+    except (TypeError, ValueError):
+        value = PLAN_MATCH_DEFAULT_THRESHOLD
+    return min(1.0, max(0.0, value))
+
+
+def plan_candidate_text(
+    description: str | None = None,
+    body: str | None = None,
+    steps: list[tuple[str | None, str | None]] | None = None,
+) -> str:
+    """What a plan that doesn't exist yet says about itself, for the gate.
+
+    The steps belong in it: a plan passed with steps and no design is still
+    recognisable by them, and what its steps say is most of what makes two
+    plans the same plan. Each step is (title, body), joined by embedding_text
+    like every other record that becomes embedded text (#2486).
+    """
+    parts = [(description or "").strip(), (body or "").strip()]
+    parts += [embeddings_svc.embedding_text(t, b) for t, b in (steps or [])]
+    return "\n\n".join(p for p in parts if p)
+
+
+async def find_matching_plan(
+    user_id: int,
+    project_id: int,
+    title: str,
+    text: str = "",
+) -> DuplicateMatch | None:
+    """An ACTIVE milestone in the project that already is this plan, or None.
+
+    Normalized-title match first, then semantic when `text` (from
+    plan_candidate_text) is long enough to mean something, the same floor the
+    note gate uses and for the same reason: a title-only embedding sits in a
+    tight neighbourhood and false-positives. Never raises; a failed check lets
+    the plan through, because a create must not depend on a recall aid.
+    """
+    from scribe.models.milestone import Milestone
+
+    if not project_id:
+        return None
+    # Rule 78, before either arm: a match names a milestone, and a caller who
+    # cannot read the project must not learn its plans by guessing titles.
+    try:
+        if not await can_read_project(user_id, project_id):
+            return None
+    except Exception:
+        logger.debug("plan gate access check failed — letting the plan through", exc_info=True)
+        return None
+    norm = " ".join((title or "").split()).lower()
+    if norm:
+        try:
+            async with async_session() as session:
+                existing = (await session.execute(
+                    select(Milestone).where(
+                        Milestone.project_id == project_id,
+                        Milestone.deleted_at.is_(None),
+                        Milestone.status == "active",
+                        func.lower(func.trim(Milestone.title)) == norm,
+                    ).limit(1)
+                )).scalars().first()
+            if existing is not None:
+                return DuplicateMatch(existing.id, existing.title, 1.0, "title")
+        except Exception:
+            logger.debug("plan gate title check skipped — query failed", exc_info=True)
+            return None
+
+    if len((text or "").strip()) < _MIN_BODY_FOR_SEMANTIC:
+        return None
+    doc_title, doc_body = embeddings_svc.milestone_document(title, None, text)
+    query = "\n\n".join(p for p in (doc_title, doc_body) if p)
+    try:
+        hits = await embeddings_svc.semantic_search_milestones(
+            user_id, query, project_id=project_id, status="active", limit=1,
+            threshold=await get_plan_match_threshold(user_id),
+        )
+    except Exception:
+        logger.debug("plan gate semantic check skipped", exc_info=True)
+        return None
+    if hits:
+        score, milestone = hits[0]
+        return DuplicateMatch(milestone.id, milestone.title, round(score, 3), "semantic")
+    return None
+
+
+def plan_match_response(dup: DuplicateMatch, progress: dict | None = None) -> dict:
+    """The payload start_planning / create_milestone return instead of a second
+    plan: the existing one, how far along it is, and how to add to it."""
+    progress = progress or {}
+    total, completed = progress.get("total", 0), progress.get("completed", 0)
+    how = "has the same title" if dup.reason == "title" else "reads as the same plan"
+    return {
+        "duplicate": True,
+        "existing_id": dup.id,
+        "existing_title": dup.title,
+        "existing_milestone": {
+            "id": dup.id,
+            "title": dup.title,
+            "description": progress.get("description") or "",
+            "total": total,
+            "completed": completed,
+        },
+        "similarity": dup.similarity,
+        "match": dup.reason,
+        "message": (
+            f'An active plan in this project {how}: milestone {dup.id} '
+            f'"{dup.title}" ({completed} of {total} steps done). Nothing was '
+            f"created. Add your steps to it with create_records(milestone_id="
+            f"{dup.id}, ...), and revise its design with update_milestone if the "
+            f"scope has grown. Read it first with get_milestone({dup.id}). If this "
+            f"really is a separate plan, retry with force=true."
+        ),
+    }
+
+
+async def plan_gate(
+    user_id: int,
+    project_id: int,
+    title: str,
+    text: str = "",
+) -> dict | None:
+    """find_matching_plan, answered: the plan_match_response to return in
+    place of a new plan, or None to go ahead and create it."""
+    from scribe.services import milestones as milestones_svc
+
+    dup = await find_matching_plan(user_id, project_id, title, text)
+    if dup is None:
+        return None
+    try:
+        rows = await milestones_svc.get_project_milestone_summary(user_id, project_id)
+        progress = next((r for r in rows if r.get("id") == dup.id), None)
+    except Exception:
+        # The match stands without its progress; losing the count must not
+        # turn a found plan into a second one.
+        progress = None
+    return plan_match_response(dup, progress)

src/scribe/services/embeddings.py

@@ -25,7 +25,8 @@ from scribe.models.embedding import NoteEmbedding, RuleEmbedding
 from scribe.models.note import Note
 from scribe.services.access import can_read_project, notes_visibility_clause
 
-if TYPE_CHECKING:  # resolves the Rule forward ref without importing at runtime
+if TYPE_CHECKING:  # resolves forward refs without importing at runtime
+    from scribe.models.milestone import Milestone
     from scribe.models.rulebook import Rule
 
 logger = logging.getLogger(__name__)
@@ -975,3 +976,169 @@ async def backfill_rule_embeddings() -> None:
     logger.info("Rule embedding backfill: embedding %d rule(s)", len(stale))
     for rule_id, title, statement, when_to_apply in stale:
         await upsert_rule_embedding(rule_id, title, statement, when_to_apply)
+
+
+# ── Milestones (milestone 415) ──────────────────────────────────────────
+
+def milestone_document(
+    title: str | None, description: str | None, body: str | None,
+) -> tuple[str | None, str | None]:
+    """The (title, body) a milestone is EMBEDDED as.
+
+    Title and one-line description lead, the way a snippet's name and purpose
+    lead its document (note 2485): the question this search answers is "does
+    a plan for this already exist?", and a plan is recognised by what it is
+    FOR. The plan body follows, chunked, so a milestone whose description is
+    empty — most roadmap milestones written by hand — is still findable by the
+    words of its design.
+    """
+    name = (title or "").strip()
+    purpose = (description or "").strip()
+    plan = (body or "").strip()
+    doc_title = f"{name} — {purpose}" if name and purpose else (name or purpose or None)
+    parts = [p for p in (purpose, plan) if p]
+    return doc_title, "\n\n".join(parts) or None
+
+
+async def upsert_milestone_embedding(
+    milestone_id: int, title: str | None, description: str | None, body: str | None,
+) -> None:
+    """Chunk, embed and persist a milestone's vectors. Safe to fire-and-forget.
+
+    The rule twin's contract: one document definition shared by the write path
+    and the backfill, and an atomic per-milestone replacement guarded by the
+    parent-row claim (#3262), so a milestone deleted mid-refresh wins.
+    """
+    from scribe.models.embedding import MilestoneEmbedding
+    from scribe.models.milestone import Milestone
+
+    doc_title, doc_body = milestone_document(title, description, body)
+    chunks = chunk_document(doc_title, doc_body)
+    try:
+        if not chunks:
+            async with async_session() as session:
+                await session.execute(
+                    delete(MilestoneEmbedding).where(MilestoneEmbedding.milestone_id == milestone_id)
+                )
+                await session.commit()
+            return
+    except Exception:
+        logger.warning("Failed to clear embedding for milestone %d", milestone_id, exc_info=True)
+        return
+
+    try:
+        vectors = await get_embeddings(chunks)
+    except Exception:
+        logger.debug("Skipping embedding for milestone %d — embedder unavailable", milestone_id)
+        return
+
+    try:
+        async with async_session() as session:
+            if not await _claim_parent_row(session, Milestone.id, milestone_id, "milestone"):
+                return
+            await session.execute(
+                delete(MilestoneEmbedding).where(MilestoneEmbedding.milestone_id == milestone_id)
+            )
+            for index, (chunk, vector) in enumerate(zip(chunks, vectors)):
+                session.add(MilestoneEmbedding(
+                    milestone_id=milestone_id, chunk_index=index, embedding=vector,
+                    chunk_text=chunk, chunker_version=CHUNKER_VERSION,
+                ))
+            await session.commit()
+    except Exception:
+        logger.warning("Failed to persist embedding for milestone %d", milestone_id, exc_info=True)
+
+
+async def semantic_search_milestones(
+    user_id: int,
+    query: str,
+    *,
+    project_id: int | None = None,
+    status: str | None = None,
+    limit: int = 5,
+    threshold: float = _SIMILARITY_THRESHOLD,
+) -> list[tuple[float, "Milestone"]]:
+    """Return up to *limit* (score, milestone) pairs most like *query*.
+
+    Answers "is there already a plan for this?" — the question a session asks
+    before start_planning, and the one the planning gate asks for it.
+
+    SCOPE. With `project_id`, that project's milestones, provided the caller
+    can read the project (access.can_read_project, rule 78) — a collaborator on
+    a shared project sees its plans. Without one, the milestones the caller
+    owns across their projects. `status` narrows to "active" or "done".
+
+    Collapses to best-chunk-per-milestone, like the sibling searches. Returns
+    an empty list if the embedder is unavailable, the project is not readable,
+    or on any error: a recall aid must never break the call it serves.
+    """
+    from scribe.models.embedding import MilestoneEmbedding
+    from scribe.models.milestone import Milestone
+
+    if not query or not query.strip():
+        return []
+    try:
+        query_vec = await get_embedding(query)
+    except Exception:
+        logger.debug("Milestone search skipped — embedder unavailable")
+        return []
+
+    distance = MilestoneEmbedding.embedding.cosine_distance(query_vec)
+    try:
+        if project_id:
+            if not await can_read_project(user_id, project_id):
+                return []
+            scope = Milestone.project_id == project_id
+        else:
+            scope = Milestone.user_id == user_id
+        async with async_session() as session:
+            rows = (await session.execute(
+                select(Milestone, distance.label("distance"))
+                .select_from(MilestoneEmbedding)
+                .join(Milestone, MilestoneEmbedding.milestone_id == Milestone.id)
+                .where(
+                    scope,
+                    Milestone.deleted_at.is_(None),
+                    *([Milestone.status == status] if status else []),
+                )
+                .order_by(distance)
+                .limit(limit * _CHUNK_OVERFETCH)
+            )).all()
+    except Exception:
+        logger.warning("Milestone semantic search failed", exc_info=True)
+        return []
+
+    best: dict[int, tuple[float, object]] = {}
+    for milestone, dist in rows:
+        score = 1.0 - float(dist)
+        if milestone.id not in best or score > best[milestone.id][0]:
+            best[milestone.id] = (score, milestone)
+    ranked = sorted(best.values(), key=lambda pair: pair[0], reverse=True)
+    return [pair for pair in ranked if pair[0] >= threshold][:limit]
+
+
+async def backfill_milestone_embeddings() -> None:
+    """Embed milestones that have no current vectors. Runs at startup beside
+    the note and rule backfills; a CHUNKER_VERSION bump re-embeds."""
+    from scribe.models.embedding import MilestoneEmbedding
+    from scribe.models.milestone import Milestone
+
+    try:
+        async with async_session() as session:
+            current = select(MilestoneEmbedding.milestone_id).where(
+                MilestoneEmbedding.chunker_version == CHUNKER_VERSION
+            )
+            stale = (await session.execute(
+                select(Milestone.id, Milestone.title, Milestone.description, Milestone.body)
+                .where(Milestone.deleted_at.is_(None), Milestone.id.notin_(current))
+            )).all()
+    except Exception:
+        logger.warning("Milestone embedding backfill: failed to query milestones", exc_info=True)
+        return
+
+    if not stale:
+        logger.info("Milestone embedding backfill: all milestones current at chunker v%d", CHUNKER_VERSION)
+        return
+    logger.info("Milestone embedding backfill: embedding %d milestone(s)", len(stale))
+    for milestone_id, title, description, body in stale:
+        await upsert_milestone_embedding(milestone_id, title, description, body)

src/scribe/services/milestones.py

@@ -11,6 +11,30 @@ from scribe.models.note import Note
 logger = logging.getLogger(__name__)
 
 
+def embed_milestone(milestone: Milestone) -> None:
+    """Refresh a milestone's vectors, fire-and-forget (milestone 415).
+
+    At the service, so every path that writes a milestone gets it — the lesson
+    embed_note records (#2056): a record written through a door that forgot the
+    call stays out of search until a restart. Exceptions are swallowed because
+    a milestone that saved must not fail on its index refresh; no running loop
+    (a script, a unit test) is ordinary. A delete racing the refresh wins: the
+    upsert claims the milestone's row first (#3262).
+    """
+    try:
+        import asyncio
+
+        from scribe.services.embeddings import upsert_milestone_embedding
+
+        asyncio.create_task(upsert_milestone_embedding(
+            milestone.id, milestone.title, milestone.description, milestone.body,
+        ))
+    except RuntimeError:
+        pass
+    except Exception:  # noqa: BLE001 - never let indexing break a write
+        logger.exception("embedding refresh failed for milestone %s", milestone.id)
+
+
 async def create_milestone(
     user_id: int,
     project_id: int,
@@ -33,6 +57,7 @@ async def create_milestone(
         session.add(milestone)
         await session.commit()
         await session.refresh(milestone)
+        embed_milestone(milestone)
         return milestone
 
 
@@ -125,6 +150,8 @@ async def update_milestone(user_id: int, milestone_id: int, **fields: object) ->
         milestone.updated_at = datetime.now(timezone.utc)
         await session.commit()
         await session.refresh(milestone)
+        if {"title", "description", "body"} & set(fields):
+            embed_milestone(milestone)
         return milestone
 
 
@@ -283,3 +310,29 @@ def brief_milestone_summary(
         )[:limit]
     brief = [{k: r[k] for k in _BRIEF_FIELDS if k in r} for r in kept]
     return brief, len(rows) - len(kept)
+
+
+def unplanned_milestones(
+    rows: list[dict], *, exclude_ids: set[int] = frozenset(), limit: int | None = None,
+) -> tuple[list[dict], int]:
+    """Active milestones with no steps yet, as (rows, omitted).
+
+    A plan written as a milestone with a description and no steps is open work
+    that nothing else names. It is never "touched" — touching is a step
+    changing — so the recency list that brief_milestone_summary(limit=) builds
+    can never reach it, and progress reads 0% either way. A project whose
+    roadmap was written that way ended up with every later plan opened as a
+    new milestone beside the one that already described it (milestone 415).
+
+    `exclude_ids` drops milestones a caller already listed. Kept in roadmap
+    order (order_index, then creation), the order they were written in.
+    Rows are id, title and description: what a reader needs to recognise the
+    plan, and not its body, which get_milestone reads.
+    """
+    found = [
+        {"id": r["id"], "title": r.get("title"), "description": r.get("description")}
+        for r in rows
+        if r.get("status") == "active" and not r.get("total") and r["id"] not in exclude_ids
+    ]
+    kept = found if limit is None else found[:limit]
+    return kept, len(found) - len(kept)

src/scribe/services/record_batch.py

@@ -28,6 +28,7 @@ from scribe.models import async_session
 from scribe.models.milestone import Milestone
 from scribe.models.note import Note
 from scribe.services import access as access_svc
+from scribe.services import milestones as milestones_svc
 from scribe.services import notes as notes_svc
 from scribe.services import systems as systems_svc
 from scribe.services.record_refs import placeholder_keys, resolve_placeholders
@@ -183,6 +184,8 @@ async def create_batch(
     # After the commit, as a single create does: embedding and System tags are
     # enrichment on records that now exist, and a failure in either must not
     # un-create them.
+    if new_ms is not None:
+        milestones_svc.embed_milestone(new_ms)
     for note, item in zip(notes, items):
         notes_svc.embed_note(note)
         if item.system_ids:

src/scribe/services/rulebooks.py

@@ -1112,36 +1112,46 @@ def rules_payload(
     (`plugin_context`) — computes a marker and shows nobody anything, and
     counting it would put rules in the denominator that no agent ever saw.
 
+    EVERY FORM LISTS, NONE RESTATES. Rules reach a session in full by
+    retrieval, so these payloads say which constraints exist — id and title,
+    the topic a global rule sits in, `via` for a co_surfaces partner — and
+    get_rule reads one. Planning reads carried the full rule_brief until a
+    project's listing grew to every global rule tagged to its areas
+    (milestone 414) and start_planning replied with 92k characters (#4081),
+    the shape #4045 had just removed from the handshake.
+
     `brief` is the session handshake's form (#4045): the project's own rules
-    as id and title, nothing else. Rules reach a session in full by
-    retrieval, so the handshake lists which of the project's constraints exist
-    rather than restating them; get_rule reads one. Only what is shown is
-    recorded as surfaced.
+    only. Only what is shown is recorded as surfaced.
     """
+    project_rules = [_rule_line(r) for r in applicable.get("project_rules", [])]
     if brief:
-        project_rules = [
-            {"id": r["id"], "title": r["title"]}
-            for r in applicable.get("project_rules", [])
-        ]
         record_rule_surfaced(
             user_id=user_id, rule_ids=[r["id"] for r in project_rules], source=source,
         )
         return {"project_rules": project_rules}
+    rules = [_rule_line(r) for r in applicable.get("rules", [])]
     record_rule_surfaced(
         user_id=user_id,
-        rule_ids=(
-            [r["id"] for r in applicable.get("rules", [])]
-            + [r["id"] for r in applicable.get("project_rules", [])]
-        ),
+        rule_ids=[r["id"] for r in rules] + [r["id"] for r in project_rules],
         source=source,
     )
     return {
-        "applicable_rules": applicable["rules"],
+        "applicable_rules": rules,
         "applicable_rules_truncated": applicable["truncated"],
-        "project_rules": applicable.get("project_rules", []),
+        "project_rules": project_rules,
     }
 
 
+def _rule_line(brief: dict) -> dict:
+    """One rule as a listing names it: enough to recognise it and fetch it.
+    Keys a row does not carry are left out rather than sent empty (#2483)."""
+    line = {"id": brief["id"], "title": brief["title"]}
+    for key in ("topic_title", "via"):
+        if brief.get(key):
+            line[key] = brief[key]
+    return line
+
+
 # ── The staleness marker (milestone 323 step 5) ────────────────────────
 #
 # WHAT THIS CAN AND CANNOT SEE. An etag catches a rule that MOVED after a

tests/conftest.py

@@ -80,7 +80,10 @@ def _no_embedding():
     """
     from unittest.mock import MagicMock
 
-    with patch("scribe.services.notes.embed_note", MagicMock()):
+    # Milestones embed too since milestone 415; a plan created in a test would
+    # otherwise detach the same model-loading task.
+    with patch("scribe.services.notes.embed_note", MagicMock()), \
+         patch("scribe.services.milestones.embed_milestone", MagicMock()):
         yield
 
 

tests/test_guidance_ownership.py

@@ -147,6 +147,12 @@ TOPICS: tuple[Topic, ...] = (
     # ── process arcs — owned by their skills ──
     Topic("plan in a milestone, steps created together", "skill:writing-plans", ("start_planning", "{{ref:"),
           "a milestone earns its place when the work has an arc", index=("start_planning",)),
+    # Milestone 415: sessions opened a second plan beside the roadmap milestone
+    # that already covered the work, because nothing told them to look.
+    Topic("find the existing plan before making one", "skill:writing-plans",
+          ('content_type="milestone"', "unplanned_milestones", "existing_milestone"),
+          "when an active milestone already covers the work, the plan is that milestone",
+          index=('search(content_type="milestone")',)),
     Topic("reuse recorded shapes; record at first build", "skill:reusing-code",
           ("create_snippet", "when_to_use", "first build", "second copy"),
           "prior art offered beside a write is not noise", index=("create_snippet",)),

tests/test_integration_milestone_search.py

@@ -0,0 +1,126 @@
+"""Real-Postgres tests for finding a plan by meaning (milestone 415, steps 3 and 4).
+
+A project's roadmap written as milestones was invisible to recall: `search`
+covered notes, tasks and rules, so "is there already a plan for this?" had no
+tool. What a mock cannot show is the join scoping the vectors to a project and
+to what the caller may read, so these seed real milestones with hand-made
+vectors and stub only the embedder.
+"""
+import uuid
+from unittest.mock import AsyncMock, patch
+
+import pytest
+import pytest_asyncio
+
+from scribe.models import async_session
+from scribe.models.embedding import EMBEDDING_DIM, MilestoneEmbedding
+from scribe.models.milestone import Milestone
+from scribe.models.project import Project
+from scribe.services import dedup as dedup_svc
+from scribe.services.embeddings import CHUNKER_VERSION, semantic_search_milestones
+from tests.helpers import ensure_user
+
+pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("_dispose_engine", "_no_embedding")]
+
+NEAR = [1.0] + [0.0] * (EMBEDDING_DIM - 1)
+FAR = [0.0, 1.0] + [0.0] * (EMBEDDING_DIM - 2)
+
+
+@pytest_asyncio.fixture
+async def roadmap():
+    tag = uuid.uuid4().hex[:8]
+    async with async_session() as s:
+        owner = await ensure_user(s, f"ms_search_owner_{tag}")
+        stranger = await ensure_user(s, f"ms_search_stranger_{tag}")
+        mine = Project(user_id=owner.id, title="Librarian")
+        other = Project(user_id=owner.id, title="Elsewhere")
+        s.add_all([mine, other])
+        await s.flush()
+        m3 = Milestone(user_id=owner.id, project_id=mine.id, title="M3 — Metadata",
+                       description="works, editions, providers, provenance", status="active")
+        done = Milestone(user_id=owner.id, project_id=mine.id, title="Covers",
+                         description="cover art", status="done")
+        unrelated = Milestone(user_id=owner.id, project_id=mine.id, title="Android client",
+                              description="native app", status="active")
+        foreign = Milestone(user_id=owner.id, project_id=other.id, title="Metadata elsewhere",
+                            description="same words, other project", status="active")
+        s.add_all([m3, done, unrelated, foreign])
+        await s.flush()
+        for ms, vec in ((m3, NEAR), (done, NEAR), (unrelated, FAR), (foreign, NEAR)):
+            s.add(MilestoneEmbedding(milestone_id=ms.id, chunk_index=0, embedding=vec,
+                                     chunk_text=ms.title, chunker_version=CHUNKER_VERSION))
+        ids = {"owner": owner.id, "stranger": stranger.id, "mine": mine.id,
+               "m3": m3.id, "done": done.id, "unrelated": unrelated.id, "foreign": foreign.id}
+        await s.commit()
+    return ids
+
+
+async def _found(user_id, **kw) -> list[int]:
+    with patch("scribe.services.embeddings.get_embedding", AsyncMock(return_value=NEAR)):
+        hits = await semantic_search_milestones(user_id, "book metadata and providers",
+                                                threshold=0.5, limit=10, **kw)
+    return [m.id for _s, m in hits]
+
+
+async def test_a_plan_is_found_in_its_project_and_not_in_another(roadmap):
+    found = await _found(roadmap["owner"], project_id=roadmap["mine"])
+    assert set(found) == {roadmap["m3"], roadmap["done"]}
+    assert roadmap["foreign"] not in found and roadmap["unrelated"] not in found
+
+
+async def test_status_narrows_to_open_plans(roadmap):
+    found = await _found(roadmap["owner"], project_id=roadmap["mine"], status="active")
+    assert found == [roadmap["m3"]]
+
+
+async def test_without_a_project_it_searches_the_callers_own(roadmap):
+    found = await _found(roadmap["owner"])
+    assert {roadmap["m3"], roadmap["done"], roadmap["foreign"]} <= set(found)
+
+
+async def test_a_project_the_caller_cannot_read_returns_nothing(roadmap):
+    assert await _found(roadmap["stranger"], project_id=roadmap["mine"]) == []
+    assert await _found(roadmap["stranger"]) == []
+
+
+# ── the plan gate (step 4): start_planning finds the plan before making another ──
+
+LONG = "Resolve works and editions against the metadata providers. " * 5
+
+
+async def _gate(roadmap, title, text="", project=None):
+    with patch("scribe.services.embeddings.get_embedding", AsyncMock(return_value=NEAR)):
+        return await dedup_svc.plan_gate(
+            roadmap["owner"], project or roadmap["mine"], title, text,
+        )
+
+
+async def test_the_gate_returns_an_active_plan_with_the_same_title(roadmap):
+    out = await _gate(roadmap, "  m3 — METADATA ")
+    assert out["duplicate"] is True and out["match"] == "title"
+    assert out["existing_milestone"]["id"] == roadmap["m3"]
+    assert f"create_records(milestone_id={roadmap['m3']}" in out["message"]
+
+
+async def test_the_gate_finds_a_plan_by_meaning_and_never_a_done_one(roadmap):
+    """NEAR matches m3 (active), `done` (done) and `foreign` (another project).
+    Only m3 is a plan this project is still working through."""
+    out = await _gate(roadmap, "Book metadata", LONG)
+    assert out["match"] == "semantic"
+    assert out["existing_id"] == roadmap["m3"]
+
+
+async def test_a_done_plan_and_another_projects_plan_do_not_block(roadmap):
+    assert await _gate(roadmap, "Covers") is None
+    assert await _gate(roadmap, "Metadata elsewhere") is None
+    async with async_session() as s:
+        m3 = await s.get(Milestone, roadmap["m3"])
+        m3.status = "done"
+        await s.commit()
+    assert await _gate(roadmap, "Book metadata", LONG) is None
+
+
+async def test_a_short_candidate_is_judged_by_title_alone(roadmap):
+    """A title-only embedding sits in a tight neighbourhood and matches
+    anything nearby, so a bare title never takes the semantic arm."""
+    assert await _gate(roadmap, "Book metadata", "just a title") is None

tests/test_mcp_tool_create_records.py

@@ -120,6 +120,7 @@ async def test_start_planning_hands_its_steps_to_the_service():
     from scribe.mcp.tools.tasks import start_planning
 
     with patch("scribe.mcp.tools.tasks.dedup_svc.find_duplicate_note", AsyncMock(return_value=None)), \
+         patch("scribe.mcp.tools.tasks.dedup_svc.plan_gate", AsyncMock(return_value=None)), \
          patch("scribe.mcp.tools.tasks.planning_svc.start_planning",
                AsyncMock(return_value={"milestone": {"id": 1}})) as svc:
         await start_planning(project_id=3, title="Plan", body="see {{ref:1}}",

tests/test_mcp_tool_milestones.py

@@ -12,6 +12,15 @@ from tests.helpers import fake_milestone
 pytestmark = pytest.mark.usefixtures("_bind_user")
 
 
+@pytest.fixture(autouse=True)
+def _no_plan_gate():
+    """The plan gate reads the database; these tests are about what reaches
+    the service. The gate's own tests re-patch it."""
+    with patch("scribe.mcp.tools.milestones.dedup_svc.plan_gate",
+               AsyncMock(return_value=None)):
+        yield
+
+
 @pytest.mark.asyncio
 async def test_list_milestones_returns_dict_with_progress():
     rows = [{"id": 1, "title": "MS1", "status": "active", "total": 2}]
@@ -63,6 +72,33 @@ async def test_create_milestone_empty_body_becomes_none():
     assert mock.call_args.kwargs["body"] is None
 
 
+@pytest.mark.asyncio
+async def test_create_milestone_returns_the_active_plan_that_already_covers_it():
+    match = {"duplicate": True, "existing_id": 4}
+    create = AsyncMock()
+    with patch("scribe.mcp.tools.milestones.dedup_svc.plan_gate",
+               AsyncMock(return_value=match)) as gate, \
+         patch("scribe.mcp.tools.milestones.milestones_svc.create_milestone", create):
+        out = await create_milestone(project_id=1, title="t", description="d", body="b")
+    assert out is match
+    create.assert_not_awaited()
+    assert gate.call_args.args[:3] == (7, 1, "t")
+    assert gate.call_args.args[3] == "d\n\nb"
+
+
+@pytest.mark.asyncio
+async def test_create_milestone_skips_the_gate_when_forced_or_done():
+    """force: the caller read the match. done: a record of past work is not a
+    plan competing with an open one."""
+    gate = AsyncMock(return_value={"duplicate": True})
+    create = AsyncMock(return_value=fake_milestone(id=6))
+    with patch("scribe.mcp.tools.milestones.dedup_svc.plan_gate", gate), \
+         patch("scribe.mcp.tools.milestones.milestones_svc.create_milestone", create):
+        assert (await create_milestone(project_id=1, title="t", force=True))["id"] == 6
+        assert (await create_milestone(project_id=1, title="t", status="done"))["id"] == 6
+    gate.assert_not_awaited()
+
+
 @pytest.mark.asyncio
 async def test_update_milestone_sends_body():
     m = fake_milestone()

tests/test_mcp_tool_planning.py

@@ -7,6 +7,13 @@ from tests.helpers import fake_task
 pytestmark = pytest.mark.usefixtures("_bind_user")
 
 
+@pytest.fixture(autouse=True)
+def _no_plan_gate():
+    """The plan gate reads the database; the tests that are about it re-patch it."""
+    with patch("scribe.mcp.tools.tasks.dedup_svc.plan_gate", AsyncMock(return_value=None)):
+        yield
+
+
 @pytest.mark.asyncio
 async def test_start_planning_tool_delegates_to_service():
     payload = {"milestone": {"id": 5}, "applicable_rules": [], "project_rules": [],
@@ -23,6 +30,42 @@ async def test_start_planning_tool_delegates_to_service():
     }
 
 
+@pytest.mark.asyncio
+async def test_start_planning_returns_the_plan_that_already_covers_it():
+    """The FabledLibrarian failure (milestone 415): a session that could not see
+    the existing plan made a second one. Now it is handed the first, and
+    nothing is created."""
+    match = {"duplicate": True, "existing_id": 12}
+    svc = AsyncMock()
+    with patch("scribe.mcp.tools.tasks.dedup_svc.plan_gate",
+               AsyncMock(return_value=match)) as gate, \
+         patch("scribe.mcp.tools.tasks.dedup_svc.find_duplicate_note", AsyncMock()) as steps, \
+         patch("scribe.mcp.tools.tasks.planning_svc.start_planning", svc):
+        from scribe.mcp.tools.tasks import start_planning
+        out = await start_planning(project_id=3, title="Metadata", body="design",
+                                   steps=[{"title": "Resolve editions", "body": "via providers"}])
+    assert out is match
+    svc.assert_not_awaited()
+    steps.assert_not_awaited()
+    # The candidate is judged by its steps as well as its design.
+    assert gate.call_args.args[:3] == (7, 3, "Metadata")
+    assert "design" in gate.call_args.args[3]
+    assert "Resolve editions\nvia providers" in gate.call_args.args[3]
+
+
+@pytest.mark.asyncio
+async def test_force_creates_the_plan_without_asking_the_gate():
+    gate = AsyncMock(return_value={"duplicate": True})
+    with patch("scribe.mcp.tools.tasks.dedup_svc.plan_gate", gate), \
+         patch("scribe.mcp.tools.tasks.planning_svc.start_planning",
+               AsyncMock(return_value={"milestone": {"id": 5}})) as svc:
+        from scribe.mcp.tools.tasks import start_planning
+        out = await start_planning(project_id=3, title="Metadata", force=True)
+    assert out["milestone"]["id"] == 5
+    svc.assert_awaited_once()
+    gate.assert_not_awaited()
+
+
 @pytest.mark.asyncio
 async def test_get_task_augments_plan_with_rules():
     applicable = {"rules": [{"id": 1, "title": "r"}], "truncated": False,

tests/test_mcp_tool_search.py

@@ -106,3 +106,34 @@ async def test_rule_search_scopes_to_the_project_it_is_given(project_id, scope):
     kwargs = found.await_args.kwargs
     assert {k: kwargs[k] for k in scope} == scope
     assert set(kwargs) & {"project_id", "everywhere"} == set(scope)
+
+
+def test_a_milestone_is_embedded_by_what_it_is_for_then_its_plan():
+    from scribe.services.embeddings import milestone_document
+
+    assert milestone_document("M3", "metadata providers", "## Goal\nx") == (
+        "M3 — metadata providers", "metadata providers\n\n## Goal\nx")
+    # A roadmap milestone written with no description is still findable by its plan.
+    assert milestone_document("M3", None, "the plan") == ("M3", "the plan")
+    assert milestone_document(None, None, None) == (None, None)
+
+
+@pytest.mark.asyncio
+async def test_milestone_search_is_its_own_shape_and_scopes_to_the_project():
+    """milestone 415: 'is there already a plan for this?' has a tool."""
+    from unittest.mock import MagicMock
+
+    _user_id_ctx.set(7)
+    ms = MagicMock(id=339, title="M3 — Metadata", description="works, editions",
+                   status="active", project_id=30)
+    found = AsyncMock(return_value=[(0.81, ms)])
+    summary = AsyncMock(return_value=[{"id": 339, "total": 0, "completed": 0}])
+    with patch("scribe.mcp.tools.search.semantic_search_milestones", found), \
+         patch("scribe.services.milestones.get_project_milestone_summary", summary):
+        out = await search(q="book metadata", content_type="milestone", project_id=30)
+    assert found.await_args.kwargs["project_id"] == 30
+    assert out["results"] == [{
+        "id": 339, "title": "M3 — Metadata", "description": "works, editions",
+        "status": "active", "project_id": 30, "total": 0, "completed": 0,
+        "similarity": 0.81,
+    }]

tests/test_milestone_summary_brief.py

@@ -14,6 +14,7 @@ import pytest
 from scribe.mcp.tools.milestones import list_milestones
 from scribe.mcp.tools.projects import enter_project, get_project
 from scribe.services.milestones import brief_milestone_summary
+from scribe.services.milestones import unplanned_milestones as brief_unplanned
 from tests.helpers import fake_project
 
 
@@ -193,3 +194,82 @@ async def test_list_milestones_lists_every_milestone_without_plans():
         out = await list_milestones(project_id=5)
     assert len(out["milestones"]) == 30
     assert all("body" not in m for m in out["milestones"])
+
+
+def test_a_planning_read_lists_rules_without_restating_them():
+    """#4081: a project's listing is every global rule tagged to its areas, so
+    the full rule_brief of each put start_planning at 92k characters. Planning
+    reads name the rules; get_rule reads one."""
+    from scribe.services.rulebooks import rules_payload
+
+    applicable = {
+        "rules": [{"id": i, "title": f"r{i}", "statement": PLAN, "when_to_apply": PLAN,
+                   "topic_title": "git", "relations": [{"note": PLAN}]} for i in range(50)]
+                 + [{"id": 900, "title": "partner", "statement": PLAN, "via": "co_surfaces"}],
+        "project_rules": [{"id": 100 + i, "title": f"pr{i}", "statement": PLAN}
+                          for i in range(30)],
+        "truncated": True,
+    }
+    with patch("scribe.services.rulebooks.record_rule_surfaced") as surfaced:
+        out = rules_payload(applicable, user_id=7, source="start_planning")
+
+    assert out["applicable_rules"][0] == {"id": 0, "title": "r0", "topic_title": "git"}
+    assert out["applicable_rules"][-1] == {"id": 900, "title": "partner", "via": "co_surfaces"}
+    assert out["project_rules"][0] == {"id": 100, "title": "pr0"}
+    assert out["applicable_rules_truncated"] is True
+    assert len(surfaced.call_args.kwargs["rule_ids"]) == 81
+    assert len(json.dumps(out)) < 6_000, len(json.dumps(out))
+
+
+# ── Milestones with no steps are open work (milestone 415) ─────────────────
+
+
+def _planless(mid: int, status: str = "active") -> dict:
+    """A roadmap milestone: a description and no steps, never touched since."""
+    row = _milestone(mid, status, touched_day=1)
+    row.update(total=0, completed=0, pct=0.0,
+               status_counts={"todo": 0, "in_progress": 0, "done": 0, "cancelled": 0})
+    return row
+
+
+def test_unplanned_lists_active_milestones_with_no_steps_in_roadmap_order():
+    rows = [_planless(3), _milestone(4, "active", 9), _planless(5, "done"), _planless(6)]
+    kept, omitted = brief_unplanned(rows)
+    assert [r["id"] for r in kept] == [3, 6]          # not the one with steps, not the done one
+    assert kept[0] == {"id": 3, "title": "M3", "description": "what M3 is for"}
+    assert omitted == 0
+
+
+def test_unplanned_respects_exclusions_and_the_cap():
+    rows = [_planless(i) for i in range(15)]
+    kept, omitted = brief_unplanned(rows, exclude_ids={0, 1}, limit=10)
+    assert [r["id"] for r in kept] == list(range(2, 12))
+    assert omitted == 3
+
+
+@pytest.mark.asyncio
+async def test_enter_project_names_a_roadmap_the_recent_list_cannot_reach():
+    """The FabledLibrarian shape: plans written as step-less milestones sat
+    beside newer milestones that did their work, and the handshake — five most
+    recently touched — could never show them."""
+    rows = _history(8) + [_planless(100), _planless(101), _planless(102, "done")]
+    out, _ = await _enter(*_enter_stubs(fake_project(id=5), rows, []))
+
+    assert [m["id"] for m in out["unplanned_milestones"]] == [100, 101]
+    listed = {m["id"] for m in out["milestone_summary"]}
+    assert not listed & {m["id"] for m in out["unplanned_milestones"]}
+    assert "unplanned_milestones_omitted" not in out
+
+
+@pytest.mark.asyncio
+async def test_no_unplanned_key_when_every_milestone_has_steps():
+    out, _ = await _enter(*_enter_stubs(fake_project(id=5), _history(4), []))
+    assert "unplanned_milestones" not in out
+
+
+@pytest.mark.asyncio
+async def test_a_long_roadmap_does_not_rebuild_the_payload():
+    rows = _history(5) + [_planless(1000 + i) for i in range(40)]
+    out, _ = await _enter(*_enter_stubs(fake_project(id=5), rows, []))
+    assert len(out["unplanned_milestones"]) == 10
+    assert out["unplanned_milestones_omitted"].startswith("30 more")

tests/test_rule_usage_wiring.py

@@ -327,8 +327,9 @@ def test_rules_payload_records_both_the_family_and_project_halves():
     with patch.object(svc, "record_rule_surfaced", rec):
         svc.rules_payload(
             {
-                "rules": [{"id": 10}, {"id": 11}],
-                "project_rules": [{"id": 12}],
+                # rule_brief always carries a title; a listing names each rule.
+                "rules": [{"id": 10, "title": "a"}, {"id": 11, "title": "b"}],
+                "project_rules": [{"id": 12, "title": "c"}],
                 "truncated": False,
             },
             user_id=1,

tests/test_services_dedup.py

@@ -4,10 +4,15 @@ from unittest.mock import AsyncMock, MagicMock, patch
 import pytest
 
 from scribe.services.dedup import (
+    PLAN_MATCH_DEFAULT_THRESHOLD,
     DuplicateMatch,
     duplicate_response,
     find_duplicate_note,
     find_duplicate_rule,
+    find_matching_plan,
+    get_plan_match_threshold,
+    plan_candidate_text,
+    plan_match_response,
 )
 from tests.helpers import fake_note, make_mock_session
 
@@ -342,3 +347,79 @@ def test_every_kind_has_a_suggestion_and_none_proposes_merging_notes():
     assert "merge" in _KIND_SUGGESTION["snippet"]
     assert "NOT merge" in _KIND_SUGGESTION["note"]
     assert "supersedes" in _KIND_SUGGESTION["note"]
+
+
+# ── the plan gate (milestone 415, step 4) ─────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_plan_title_match_short_circuits_the_semantic_arm():
+    ms = MagicMock(id=415, title="Plan gate")
+    sem = AsyncMock()
+    with patch("scribe.services.dedup.can_read_project", AsyncMock(return_value=True)), \
+         patch("scribe.services.dedup.async_session", return_value=_session_returning(ms)), \
+         patch("scribe.services.dedup.embeddings_svc.semantic_search_milestones", sem):
+        dup = await find_matching_plan(7, 2, " plan  GATE", "x" * 300)
+    assert (dup.id, dup.reason, dup.similarity) == (415, "title", 1.0)
+    sem.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_plan_semantic_arm_asks_for_active_plans_in_the_project_at_the_setting():
+    ms = MagicMock(id=9, title="Metadata")
+    sem = AsyncMock(return_value=[(0.912345, ms)])
+    with patch("scribe.services.dedup.can_read_project", AsyncMock(return_value=True)), \
+         patch("scribe.services.dedup.async_session", return_value=_session_returning(None)), \
+         patch("scribe.services.dedup.embeddings_svc.semantic_search_milestones", sem), \
+         patch("scribe.services.settings.get_setting", AsyncMock(return_value="0.8")):
+        dup = await find_matching_plan(7, 2, "Book metadata", "x" * 300)
+    assert (dup.id, dup.reason, dup.similarity) == (9, "semantic", 0.912)
+    kw = sem.call_args.kwargs
+    assert (kw["project_id"], kw["status"], kw["threshold"]) == (2, "active", 0.8)
+
+
+@pytest.mark.asyncio
+async def test_plan_gate_fails_open():
+    boom = MagicMock(side_effect=RuntimeError("db down"))
+    with patch("scribe.services.dedup.can_read_project", AsyncMock(return_value=True)), \
+         patch("scribe.services.dedup.async_session", boom):
+        assert await find_matching_plan(7, 2, "Anything", "x" * 300) is None
+    with patch("scribe.services.dedup.can_read_project", AsyncMock(side_effect=RuntimeError)):
+        assert await find_matching_plan(7, 2, "Anything", "x" * 300) is None
+    assert await find_matching_plan(7, 0, "No project") is None
+
+
+@pytest.mark.asyncio
+async def test_plan_gate_says_nothing_about_a_project_the_caller_cannot_read():
+    ms = MagicMock(id=415, title="Their plan")
+    session = MagicMock(return_value=_session_returning(ms))
+    with patch("scribe.services.dedup.can_read_project", AsyncMock(return_value=False)), \
+         patch("scribe.services.dedup.async_session", session):
+        assert await find_matching_plan(8, 2, "Their plan") is None
+    session.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_a_bad_threshold_setting_falls_back_to_the_default():
+    with patch("scribe.services.settings.get_setting", AsyncMock(return_value="lots")):
+        assert await get_plan_match_threshold(7) == PLAN_MATCH_DEFAULT_THRESHOLD
+    with patch("scribe.services.settings.get_setting", AsyncMock(return_value="7")):
+        assert await get_plan_match_threshold(7) == 1.0
+
+
+def test_plan_candidate_text_carries_the_steps():
+    text = plan_candidate_text(description=None, body="  design ",
+                               steps=[("Step one", None), ("Step two", "with a body"), (None, None)])
+    assert text == "design\n\nStep one\n\nStep two\nwith a body"
+
+
+def test_plan_match_response_points_at_adding_steps_not_a_second_plan():
+    out = plan_match_response(DuplicateMatch(12, "Metadata", 0.93, "semantic"),
+                              {"total": 5, "completed": 2, "description": "providers"})
+    assert out["duplicate"] is True and out["existing_id"] == 12
+    assert out["existing_milestone"] == {
+        "id": 12, "title": "Metadata", "description": "providers", "total": 5, "completed": 2,
+    }
+    for phrase in ("create_records(milestone_id=12", "get_milestone(12)", "force=true",
+                   "2 of 5 steps"):
+        assert phrase in out["message"]

tests/test_settings_defaults_agree.py

@@ -36,28 +36,32 @@ import re
 import pytest
 
 ROOT = pathlib.Path(__file__).resolve().parents[1]
-_PY = ROOT / "src" / "scribe" / "services" / "plugin_context.py"
+_SERVICES = ROOT / "src" / "scribe" / "services"
 _VUE = ROOT / "frontend" / "src" / "views" / "SettingsView.vue"
 
-# (python constant, vue ref). Hand-written because the pairing is an editorial
-# fact — the names do not share a convention either side could derive — but
-# every entry is asserted to EXIST on both sides, so a rename fails loudly
-# here rather than silently dropping that threshold from the check.
+# (services module, python constant, vue ref). Hand-written because the
+# pairing is an editorial fact — the names do not share a convention either
+# side could derive — but every entry is asserted to EXIST on both sides, so a
+# rename fails loudly here rather than silently dropping that threshold from
+# the check.
 _PAIRS = (
-    ("AUTOINJECT_DEFAULT_THRESHOLD", "kbInjectThreshold"),
-    ("WRITEPATH_DEFAULT_THRESHOLD", "kbWritePathThreshold"),
-    ("RULEHINT_DEFAULT_THRESHOLD", "kbRuleHintThreshold"),
-    ("TOOLRULE_DEFAULT_THRESHOLD", "kbToolRuleThreshold"),
-    ("PROMPTRULE_DEFAULT_THRESHOLD", "kbPromptRuleThreshold"),
+    ("plugin_context.py", "AUTOINJECT_DEFAULT_THRESHOLD", "kbInjectThreshold"),
+    ("plugin_context.py", "WRITEPATH_DEFAULT_THRESHOLD", "kbWritePathThreshold"),
+    ("plugin_context.py", "RULEHINT_DEFAULT_THRESHOLD", "kbRuleHintThreshold"),
+    ("plugin_context.py", "TOOLRULE_DEFAULT_THRESHOLD", "kbToolRuleThreshold"),
+    ("plugin_context.py", "PROMPTRULE_DEFAULT_THRESHOLD", "kbPromptRuleThreshold"),
+    # The plan gate (milestone 415): it blocks a create, so a form showing a
+    # looser bar than the one in force would be the more misleading drift.
+    ("dedup.py", "PLAN_MATCH_DEFAULT_THRESHOLD", "kbPlanMatchThreshold"),
 )
 
 
-def _python_default(name: str) -> float:
+def _python_default(module: str, name: str) -> float:
     m = re.search(rf"^{re.escape(name)}\s*=\s*([0-9.]+)\s*$",
-                  _PY.read_text(), re.M)
+                  (_SERVICES / module).read_text(), re.M)
     assert m, (
         f"{name} is no longer a bare module-level float in "
-        f"services/plugin_context.py. If it moved or was renamed, update "
+        f"services/{module}. If it moved or was renamed, update "
         f"_PAIRS; if it was retired, drop its row — leaving it here checks "
         f"nothing while looking like coverage."
     )
@@ -75,11 +79,11 @@ def _vue_default(ref_name: str) -> float:
     return float(m.group(1))
 
 
-@pytest.mark.parametrize(("constant", "ref_name"), _PAIRS,
-                         ids=[p[0] for p in _PAIRS])
-def test_the_form_shows_the_default_the_server_uses(constant, ref_name):
+@pytest.mark.parametrize(("module", "constant", "ref_name"), _PAIRS,
+                         ids=[p[1] for p in _PAIRS])
+def test_the_form_shows_the_default_the_server_uses(module, constant, ref_name):
     """An untouched control must render the bar actually in force."""
-    server, form = _python_default(constant), _vue_default(ref_name)
+    server, form = _python_default(module, constant), _vue_default(ref_name)
     assert form == server, (
         f"SettingsView shows {form} for {ref_name} while the server defaults "
         f"to {server} ({constant}). An operator who has never set this reads "