feat(plans): הרשאות-סוכנים + dedup/merge/edit + API לתור-אישור תכניות (backend)

Follow-ups למרשם-התכניות (PR #252), חלק ה-backend (ללא-UI):
- סוכנים: הוספת extract_plans/plan_get/plan_search/plan_list ל-CEO ול-חוקר
  (+ plan_upsert לחוקר); plan_review נשאר אנושי בלבד (G10). .claude/agents/*.md.
- db: _plan_core_token + find_similar_plans (הצפת כפילות-וריאנט לאישור ידני, בלי
  מיזוג-אוטומטי), update_plan (עריכה+renumber, guard התנגשות→merge), merge_plans
  (איחוד aliases, מילוי-חוסר, סתירות→discrepancies, מחיקת מקור).
- plans_extractor: צירוף possible_duplicates לפלט החילוץ.
- web/app.py: GET /api/plans(+/{id},/{id}/duplicates) · POST /api/plans · PATCH
  /api/plans/{id} · POST /api/plans/{id}/review · POST /api/plans/merge; +קטגוריית
  "תכניות הממתינות לאישור" ב-/api/chair/pending.

תיקון-נתונים (DB, מחוץ ל-PR): הל/מח/250 ד' → 7.1.2002, י"פ 5045.
ה-UI (טאב /precedents + מונה /approvals) ב-PR נפרד אחרי שער Claude Design.

Invariants: G1 (נרמול בכתיבה/עריכה) · G2 · G3 · G10 (review_status + מיזוג ידני,
אישור אנושי) · INV-DM2/DM5 · INV-AH · X9 · אין בליעה שקטה (discrepancies).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

mcp-server/src/legal_mcp/services/db.py

@@ -3698,6 +3698,163 @@ async def set_plan_review_status(plan_id: UUID, status: str) -> dict | None:
     return _plan_row_to_dict(row)
 
 
+# words that describe a plan but aren't its identifying code — stripped for the
+# comparable "core token" used in near-duplicate detection (NOT for the key).
+_PLAN_DESC_WORDS = re.compile(
+    r"(?:תכנית|תוכנית|מתאר|מקומית|מחוזית|ארצית|מפורטת|כוללנית|מספר|מס['\"׳״]?)"
+)
+
+
+def _plan_core_token(s: str) -> str:
+    """Best-effort comparable 'core' of a plan identifier — for near-duplicate
+    surfacing only, NEVER a key (bare numbers collide). Strips descriptive words
+    (תכנית/מתאר/מקומית/מס') and quotes, keeps the alphanumeric code:
+      "תכנית מתאר מקומית מס' 62" → "62" · "מי/820" → "מי/820" · "5166/ב" → "5166/ב".
+    """
+    t = _normalize_plan_number(s)
+    t = _PLAN_DESC_WORDS.sub(" ", t)
+    t = t.replace('"', " ").replace("'", " ").replace("׳", " ").replace("״", " ")
+    return " ".join(t.split()).strip()
+
+
+async def find_similar_plans(
+    plan_number: str,
+    display_name: str = "",
+    exclude_id: UUID | None = None,
+    limit: int = 10,
+) -> list[dict]:
+    """Surface plans that LOOK like the same scheme written differently — so the
+    chair can merge them (manual gate, G10). Does NOT merge. Each hit carries a
+    `match_reason`. The registry is small (dozens), so we scan and score in Python.
+    """
+    num = _normalize_plan_number(plan_number)
+    core = _plan_core_token(plan_number) or _plan_core_token(display_name)
+    if not num and not core:
+        return []
+    pool = await get_pool()
+    rows = await pool.fetch("SELECT * FROM plans")
+    out: list[dict] = []
+    exclude = str(exclude_id) if exclude_id else None
+    for r in rows:
+        d = _plan_row_to_dict(r)
+        if exclude and d["id"] == exclude:
+            continue
+        cand_num = d["plan_number"]
+        reason = None
+        if num and cand_num and (num == cand_num or num in cand_num or cand_num in num):
+            reason = "מספר חופף"
+        elif num and num in (d.get("aliases") or []):
+            reason = "alias תואם"
+        elif core and core == _plan_core_token(cand_num):
+            reason = "קוד-ליבה זהה"
+        if reason:
+            out.append({**d, "match_reason": reason})
+        if len(out) >= limit:
+            break
+    return out
+
+
+async def update_plan(
+    plan_id: UUID,
+    plan_number: str,
+    display_name: str = "",
+    plan_type: str = "",
+    gazette_date=None,
+    yalkut_number: str = "",
+    purpose: str = "",
+    aliases: list[str] | None = None,
+) -> dict | None:
+    """Edit a plan by id (chair UI). Re-normalizes plan_number (G1) and re-renders
+    the citation. Refuses to collide with another row's number — that's a merge, not
+    an edit (no silent dup). `aliases=None` keeps existing aliases.
+    """
+    num = _normalize_plan_number(plan_number)
+    if not num:
+        raise ValueError("plan_number ריק לאחר נרמול")
+    gd = _coerce_plan_date(gazette_date)
+    display_name = (display_name or "").strip() or num
+    citation = format_plan_citation({
+        "display_name": display_name, "plan_number": num,
+        "gazette_date": gd, "yalkut_number": yalkut_number, "purpose": purpose,
+    })
+    pool = await get_pool()
+    async with pool.acquire() as conn:
+        async with conn.transaction():
+            clash = await conn.fetchrow(
+                "SELECT id FROM plans WHERE plan_number = $1 AND id <> $2", num, plan_id,
+            )
+            if clash:
+                raise ValueError(
+                    f"מספר-תכנית {num} כבר קיים ברשומה אחרת — השתמש במיזוג (merge), לא בעריכה"
+                )
+            row = await conn.fetchrow(
+                """UPDATE plans SET
+                       plan_number = $2, display_name = $3, plan_type = $4,
+                       gazette_date = $5, yalkut_number = $6, purpose = $7,
+                       citation_formatted = $8,
+                       aliases = COALESCE($9, aliases), updated_at = now()
+                   WHERE id = $1 RETURNING *""",
+                plan_id, num, display_name, plan_type or "", gd,
+                yalkut_number or "", purpose or "", citation,
+                sorted(set(aliases)) if aliases is not None else None,
+            )
+    return _plan_row_to_dict(row)
+
+
+async def merge_plans(source_id: UUID, target_id: UUID) -> dict:
+    """Fold `source` into `target` (chair-initiated): union aliases (+ source's
+    number/name as aliases), fill empty target validity from source, record any
+    CONFLICTING validity in target.discrepancies (no silent loss), delete source.
+    Returns the merged target.
+    """
+    if source_id == target_id:
+        raise ValueError("מקור ויעד זהים")
+    pool = await get_pool()
+    async with pool.acquire() as conn:
+        async with conn.transaction():
+            src = await conn.fetchrow("SELECT * FROM plans WHERE id = $1 FOR UPDATE", source_id)
+            tgt = await conn.fetchrow("SELECT * FROM plans WHERE id = $1 FOR UPDATE", target_id)
+            if not src or not tgt:
+                raise ValueError("מקור או יעד לא נמצאו")
+            src, tgt = dict(src), dict(tgt)
+
+            aliases = set(tgt.get("aliases") or []) | set(src.get("aliases") or [])
+            for a in (src["plan_number"], src["display_name"]):
+                if a and a not in (tgt["plan_number"], tgt["display_name"]):
+                    aliases.add(a)
+
+            disc = tgt.get("discrepancies") or []
+            if isinstance(disc, str):
+                disc = json.loads(disc)
+            for field in ("gazette_date", "yalkut_number", "purpose"):
+                sv, tv = src[field], tgt[field]
+                if sv and tv and str(sv) != str(tv):
+                    disc.append({
+                        "field": field, "old": str(tv), "new": str(sv),
+                        "source_case_number": src.get("source_case_number") or "",
+                        "via": "merge",
+                    })
+
+            new_gd = tgt["gazette_date"] or src["gazette_date"]
+            new_yalkut = tgt["yalkut_number"] or src["yalkut_number"]
+            new_purpose = tgt["purpose"] or src["purpose"]
+            new_type = tgt["plan_type"] or src["plan_type"]
+            citation = format_plan_citation({
+                "display_name": tgt["display_name"], "plan_number": tgt["plan_number"],
+                "gazette_date": new_gd, "yalkut_number": new_yalkut, "purpose": new_purpose,
+            })
+            row = await conn.fetchrow(
+                """UPDATE plans SET aliases = $2, gazette_date = $3, yalkut_number = $4,
+                       purpose = $5, plan_type = $6, citation_formatted = $7,
+                       discrepancies = $8, updated_at = now()
+                   WHERE id = $1 RETURNING *""",
+                target_id, sorted(aliases), new_gd, new_yalkut, new_purpose,
+                new_type, citation, json.dumps(disc, ensure_ascii=False),
+            )
+            await conn.execute("DELETE FROM plans WHERE id = $1", source_id)
+    return _plan_row_to_dict(row)
+
+
 # ── V7: External precedent library + halachot ─────────────────────