feat(legal-aid): async parse jobs to survive multi-minute vision latency

Claude Vision via ai-gateway's claude-code OAuth provider runs ~140s per
rendered page (a 3-page report measured at 417s), which blows past any safe
synchronous HTTP timeout — the EspoCRM controller's 240s curl gave up with
"0 bytes received" while ai-gateway kept working.

parse-pdf-report now accepts ?mode=async: it returns {jobId, status:"pending"}
immediately and runs the parse in the background, persisting the result/error
to a file-based job store under /opt/data (survives multi-worker + restarts).
Poll GET /parse-pdf-report/jobs/{jobId}. Default sync mode is unchanged so
n8n and other server-side callers keep working.

Refs prod ai-gateway log id cee4c17c (claudeMs=417513).

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

.gitea/workflows/deploy.yaml

@@ -48,8 +48,14 @@ jobs:
             docker push "${BASE}:${VERSION}"
           fi
 
-      - name: Trigger Coolify redeploy
+      - name: Trigger Coolify redeploy (dev)
         run: |
           curl -sf \
             "http://coolify:8080/api/v1/deploy?uuid=${{ env.COOLIFY_UUID }}&force=true" \
             -H "Authorization: Bearer ${{ secrets.COOLIFY_TOKEN }}"
+
+      - name: Trigger Coolify redeploy (prod)
+        run: |
+          curl -sf \
+            "https://coolify.prod.marcus-law.co.il/api/v1/deploy?uuid=${{ secrets.COOLIFY_PROD_UUID }}&force=true" \
+            -H "Authorization: Bearer ${{ secrets.COOLIFY_PROD_TOKEN }}"

.taskmaster/config.json

@@ -1,20 +1,20 @@
 {
   "models": {
     "main": {
-      "provider": "anthropic",
-      "modelId": "claude-sonnet-4-20250514",
+      "provider": "claude-code",
+      "modelId": "sonnet",
       "maxTokens": 64000,
       "temperature": 0.2
     },
     "research": {
-      "provider": "perplexity",
-      "modelId": "sonar",
+      "provider": "claude-code",
+      "modelId": "sonnet",
       "maxTokens": 8700,
       "temperature": 0.1
     },
     "fallback": {
-      "provider": "anthropic",
-      "modelId": "claude-3-7-sonnet-20250219",
+      "provider": "claude-code",
+      "modelId": "sonnet",
       "maxTokens": 120000,
       "temperature": 0.2
     }

api/__init__.py

@@ -0,0 +1,27 @@
+"""shira-hermes API package."""
+
+from __future__ import annotations
+
+import tomllib
+from pathlib import Path
+
+
+def _read_version() -> str:
+    """Source-of-truth version from pyproject.toml.
+
+    pyproject lives at the repo root; we walk up from this file
+    instead of relying on importlib.metadata so the version is
+    correct in editable installs and in the Docker image where the
+    package may not be `pip install`-ed at all.
+    """
+    here = Path(__file__).resolve()
+    for parent in [here.parent, *here.parents]:
+        candidate = parent / "pyproject.toml"
+        if candidate.exists():
+            with candidate.open("rb") as f:
+                data = tomllib.load(f)
+            return data.get("project", {}).get("version", "0.0.0")
+    return "0.0.0"
+
+
+__version__ = _read_version()

api/main.py

@@ -9,10 +9,13 @@ from pathlib import Path
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 
+from api import __version__
 from api.routes.admin_kb import router as admin_kb_router
 from api.routes.health import router as health_router
 from api.routes.kb_public import router as kb_public_router
+from api.routes.legal_aid import router as legal_aid_router
 from api.routes.smart_assistant import router as smart_assistant_router
+from api.services.skills import seed_skills_from_image
 
 logging.basicConfig(
     level=logging.INFO,
@@ -24,7 +27,7 @@ logger = logging.getLogger("shira.app")
 app = FastAPI(
     title="shira-hermes",
     description="Shira AI Assistant — Hermes Agent backend for EspoCRM SmartAssistant",
-    version="0.2.0",
+    version=__version__,
 )
 
 app.add_middleware(
@@ -38,11 +41,12 @@ app.include_router(health_router)
 app.include_router(smart_assistant_router)
 app.include_router(admin_kb_router)
 app.include_router(kb_public_router)
+app.include_router(legal_aid_router)
 
 
 @app.on_event("startup")
 async def ensure_data_dirs():
-    """Create persistent data directories if they don't exist."""
+    """Create persistent data directories and seed any new baked-in skills."""
     data_dir = os.environ.get("SHIRA_DATA_DIR", "/opt/data")
     dirs = [
         f"{data_dir}/skills",
@@ -52,3 +56,6 @@ async def ensure_data_dirs():
     for d in dirs:
         Path(d).mkdir(parents=True, exist_ok=True)
     logger.info("Data directories ready at %s", data_dir)
+
+    added, skipped = seed_skills_from_image()
+    logger.info("Skills seed: %d new baked skill(s) added, %d already present", added, skipped)

api/routes/admin_kb.py

@@ -380,6 +380,13 @@ async def delete_admin_source(request: Request, source_id: int):
 
 @router.post("/sources/{source_id}/reingest")
 async def reingest_admin_source(request: Request, source_id: int):
+    """Kick off a two-phase re-ingest: AI re-classifies the existing
+    file, the user reviews suggestions on the batch review screen, and
+    the commit overwrites kb_source metadata + replaces chunks.
+
+    Returns `batch_id` so the browser can open the same review UI as
+    a fresh upload (single-file batch).
+    """
     _verify_admin(request)
     requested_by = (
         request.headers.get("X-User-Name")
@@ -387,14 +394,21 @@ async def reingest_admin_source(request: Request, source_id: int):
         or None
     )
     try:
-        job_id = await kb_admin_sources.start_reingest(source_id, requested_by)
+        result = await kb_admin_sources.start_reingest(source_id, requested_by)
     except LookupError:
         raise HTTPException(status_code=404, detail="Source not found")
     except ValueError as e:
         raise HTTPException(status_code=400, detail=str(e))
 
-    asyncio.create_task(kb_admin_sources.process_reingest_job(job_id))
-    return {"job_id": job_id, "status": "queued", "source_id": source_id}
+    # Schedule classify (fail-soft) — same path as fresh uploads, so
+    # the user gets fresh AI metadata suggestions on the review screen.
+    asyncio.create_task(kb_jobs.process_classify_stage(result["job_id"]))
+    return {
+        "job_id": result["job_id"],
+        "batch_id": result["batch_id"],
+        "status": "queued",
+        "source_id": source_id,
+    }
 
 
 # ── Phase 4: topic CRUD (Task #15) ─────────────────────────────────────────

api/routes/health.py

@@ -6,9 +6,9 @@ from datetime import datetime, timezone
 
 from fastapi import APIRouter
 
-router = APIRouter()
+from api import __version__
 
-VERSION = "0.2.0"
+router = APIRouter()
 
 
 @router.get("/api/health")
@@ -16,6 +16,6 @@ async def health():
     return {
         "status": "ok",
         "service": "shira-hermes",
-        "version": VERSION,
+        "version": __version__,
         "timestamp": datetime.now(timezone.utc).isoformat(),
     }

api/routes/legal_aid.py

@@ -0,0 +1,450 @@
+"""Legal Aid PDF parser endpoint.
+
+Parses a monthly Legal Aid (סיוע משפטי) payment-report PDF and returns the
+JSON shape that GreenInvoiceBilling's LegalAidPaymentReportService.ingest()
+expects.
+
+v3 (2026-05-03): switched extraction from pdfplumber to Claude Vision via
+ai-gateway. The Legal Aid PDF format has multi-line cells, RTL Hebrew column
+ordering quirks, and amounts that wrap across rows ("1478.5\\n4" → 1478.54).
+pdfplumber's table detection couldn't get past those cleanly. Claude reads
+the rendered page image natively and returns structured rows.
+
+Pattern mirrors api/services/ocr.py — render each PDF page with PyMuPDF,
+send the PNG to ai-gateway with a JSON-output prompt, then assemble the
+result. Auth is X-Api-Key as for the other admin endpoints.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import base64
+import io
+import json
+import logging
+import os
+import re
+import time
+import uuid
+from typing import Any
+
+from fastapi import APIRouter, File, HTTPException, Query, Request, UploadFile
+
+logger = logging.getLogger("shira.legal_aid")
+
+router = APIRouter(prefix="/api/legal-aid", tags=["legal-aid"])
+
+
+def _verify_admin(request: Request) -> None:
+    expected = os.environ.get("ADMIN_API_KEY") or os.environ.get("API_KEY", "")
+    if not expected:
+        raise HTTPException(status_code=503, detail="Admin API key not configured")
+    provided = request.headers.get("X-Admin-Key") or request.headers.get("X-Api-Key") or ""
+    if provided != expected:
+        raise HTTPException(status_code=401, detail="Unauthorized")
+
+
+_PDF_RENDER_DPI = 180
+_PDF_MAX_PAGES = 8  # safety cap; real reports are 1-2 pages
+
+# --- Async job store ---------------------------------------------------
+# Claude Vision via ai-gateway's claude-code OAuth provider runs ~140s per
+# rendered page (a 3-page report measured at 417s). That far exceeds the
+# EspoCRM controller's synchronous curl timeout, so manual uploads run as a
+# background job: POST ?mode=async returns a jobId immediately and the caller
+# polls GET /parse-pdf-report/jobs/{jobId}. Jobs are persisted as files under
+# /opt/data (a named volume) so they survive across uvicorn workers and short
+# restarts. n8n / other server-side callers can still use the default sync mode.
+_JOBS_DIR = os.environ.get("LEGAL_AID_JOBS_DIR", "/opt/data/legal-aid-jobs")
+_JOB_TTL_SECONDS = 3600  # finished jobs are reaped after 1h
+_running_jobs: set[asyncio.Task] = set()
+
+
+def _jobs_dir() -> str:
+    os.makedirs(_JOBS_DIR, exist_ok=True)
+    return _JOBS_DIR
+
+
+def _job_path(job_id: str) -> str:
+    # job_id is a server-generated uuid hex; sanitize defensively anyway so a
+    # crafted id from the status route can never escape the jobs directory.
+    safe = re.sub(r"[^a-f0-9]", "", job_id.lower())
+    if not safe:
+        safe = "_invalid_"
+    return os.path.join(_jobs_dir(), f"{safe}.json")
+
+
+def _write_job(job_id: str, data: dict[str, Any]) -> None:
+    path = _job_path(job_id)
+    tmp = f"{path}.tmp"
+    with open(tmp, "w", encoding="utf-8") as f:
+        json.dump(data, f, ensure_ascii=False)
+    os.replace(tmp, path)  # atomic publish
+
+
+def _read_job(job_id: str) -> dict[str, Any] | None:
+    path = _job_path(job_id)
+    if not os.path.exists(path):
+        return None
+    try:
+        with open(path, encoding="utf-8") as f:
+            return json.load(f)
+    except (OSError, ValueError):
+        return None
+
+
+def _cleanup_old_jobs() -> None:
+    try:
+        now = time.time()
+        for name in os.listdir(_jobs_dir()):
+            if not name.endswith(".json"):
+                continue
+            p = os.path.join(_JOBS_DIR, name)
+            try:
+                if now - os.path.getmtime(p) > _JOB_TTL_SECONDS:
+                    os.remove(p)
+            except OSError:
+                pass
+    except OSError:
+        pass
+
+
+_EXTRACT_PROMPT = """\
+אתה מחלץ נתונים מדוח תשלומים חודשי של הסיוע המשפטי (Legal Aid, עברית RTL).
+
+תוציא JSON תקני בלבד (בלי backticks, בלי טקסט נוסף) בסכמה הזו:
+{
+  "issuerOffice": "Jerusalem" | "TelAviv" | "Haifa" | "BeerSheva" | "Nazareth" | "Center" | "National" | null,
+  "reportDate": "YYYY-MM-DD" | null,
+  "reportMonth": "YYYY-MM" | null,
+  "totalExclVat": number,
+  "totalVat": number,
+  "totalInclVat": number,
+  "lines": [
+    {
+      "invoiceNumber": "<16 ספרות מתוך 'מספר חשבונית'>",
+      "requestNumber": "<7 ספרות מתוך 'מספר בקשה'>",
+      "legalAidCaseNumber": "<7 ספרות מתוך 'מספר תיק'>",
+      "clientName": "<שם לקוח בעברית, מאוחד אם נשבר לשני שורות>",
+      "activityCategory": "<עמודת 'פעולה' — קטגוריית האב, למשל: הודעה, לימוד ענין, התיעצות / פגישה, בקשה, דיון>",
+      "activitySubTypeHebrew": "<עמודת 'תת סוג פעולה' — תווית מלאה, למשל: הודעה בהתאם להחלטה, לימוד ענין מלא, התייעצות עם מסמכים>",
+      "activityDate": "<עמודת 'תאריך פעילות' בפורמט YYYY-MM-DD>",
+      "amountExclVat": <עמודת 'סכום' (ללא מע\"מ) כמספר עשרוני>,
+      "vat": <עמודת 'מע\"מ' כמספר עשרוני>,
+      "amountInclVat": <עמודת 'סה\"כ' כמספר עשרוני>
+    }
+  ],
+  "travelReimbursementLines": []
+}
+
+הנחיות חשובות:
+- הטבלה הראשית בעמוד 1 היא "פעולות חייבות במע\"מ" — חלץ ממנה את כל השורות.
+- אם תא בטבלה משתרע על שתי שורות (כמו "הודעה בהתאם" + "להחלטה"), אחד אותם לטקסט אחד עם רווח.
+- אם סכום נשבר לשורות (כמו "1478.5" ו-"4"), אחד אותם ל-1478.54.
+- "מספר חשבונית" הוא תמיד 16 ספרות; "מספר בקשה" ו-"מספר תיק" 7 ספרות כל אחד.
+- שדה reportMonth מתוך "עבור חודש: M/YYYY" → "YYYY-MM".
+- שדה reportDate מתוך התאריך בכותרת הראשית של הדוח (DD/MM/YYYY → YYYY-MM-DD).
+- אם אין שדה — null. אם הטבלה ריקה (כמו טבלת נסיעות בעמוד 2) — אל תכניס את שורותיה ל-lines.
+- ה-issuerOffice לפי "מחוז ירושלים" → "Jerusalem" וכו'.
+
+החזר JSON בלבד.
+"""
+
+
+async def _parse_pdf_to_payload(
+    pdf_bytes: bytes, subject: str | None, debug: bool
+) -> dict[str, Any]:
+    """Render the PDF pages and run Claude Vision to produce the ingest payload.
+
+    Shared by both the synchronous route and the async job runner. Raises
+    HTTPException on any failure (the job runner converts those to a stored
+    error status).
+    """
+    if not pdf_bytes:
+        raise HTTPException(status_code=400, detail="Empty file")
+    if pdf_bytes[:4] != b"%PDF":
+        raise HTTPException(status_code=400, detail="File is not a PDF")
+
+    try:
+        import fitz  # PyMuPDF
+    except ImportError as e:
+        raise HTTPException(status_code=503, detail=f"PyMuPDF not installed: {e}")
+
+    images_b64: list[tuple[str, str]] = []
+    page_count = 0
+    try:
+        with fitz.open(stream=pdf_bytes, filetype="pdf") as doc:
+            page_count = min(len(doc), _PDF_MAX_PAGES)
+            for i in range(page_count):
+                pix = doc[i].get_pixmap(dpi=_PDF_RENDER_DPI)
+                # JPEG (not PNG): a rendered table page is mostly white, and
+                # PNG of it is several × larger than a q85 JPEG. Bundling all
+                # pages as PNG blew past ai-gateway's body limit (413). JPEG
+                # q85 keeps the table text crisp enough for Claude Vision.
+                jpg = pix.tobytes("jpeg", jpg_quality=85)
+                images_b64.append(("image/jpeg", base64.b64encode(jpg).decode("ascii")))
+    except Exception as e:
+        logger.exception("Legal Aid PDF render failed")
+        raise HTTPException(status_code=422, detail=f"PDF render error: {e}")
+
+    if not images_b64:
+        raise HTTPException(status_code=422, detail="PDF contained no pages")
+
+    raw_response = await _call_vision_for_json(images_b64, _EXTRACT_PROMPT)
+    parsed = _parse_json_response(raw_response)
+
+    sysref = None
+    if subject:
+        m = re.search(r"\[סימוכין\s+(\d+)\]", subject)
+        if m:
+            sysref = m.group(1)
+
+    out: dict[str, Any] = {
+        "subject": subject,
+        "sysref": sysref,
+        "issuerOffice": parsed.get("issuerOffice"),
+        "reportDate": parsed.get("reportDate"),
+        "reportMonth": parsed.get("reportMonth"),
+        "totalExclVat": _to_float(parsed.get("totalExclVat")),
+        "totalVat": _to_float(parsed.get("totalVat")),
+        "totalInclVat": _to_float(parsed.get("totalInclVat")),
+        "lines": _normalize_lines(parsed.get("lines") or []),
+        "travelReimbursementLines": parsed.get("travelReimbursementLines") or [],
+    }
+
+    if debug:
+        out["_debug"] = {
+            "pageCount": page_count,
+            "rawResponse": raw_response,
+        }
+
+    return out
+
+
+async def _run_parse_job(
+    job_id: str, pdf_bytes: bytes, subject: str | None, debug: bool
+) -> None:
+    """Background runner: parse the PDF and persist the result/error to disk."""
+    try:
+        result = await _parse_pdf_to_payload(pdf_bytes, subject, debug)
+        _write_job(
+            job_id,
+            {"jobId": job_id, "status": "done", "result": result, "finishedAt": time.time()},
+        )
+    except HTTPException as e:
+        _write_job(
+            job_id,
+            {"jobId": job_id, "status": "error", "detail": str(e.detail), "finishedAt": time.time()},
+        )
+    except Exception as e:  # noqa: BLE001 — anything unexpected becomes a stored error
+        logger.exception("Legal Aid async parse job %s failed", job_id)
+        _write_job(
+            job_id,
+            {"jobId": job_id, "status": "error", "detail": str(e), "finishedAt": time.time()},
+        )
+
+
+@router.post("/parse-pdf-report")
+async def parse_pdf_report(
+    request: Request,
+    file: UploadFile = File(..., description="Legal Aid monthly payment report PDF"),
+    subject: str | None = None,
+    mode: str = Query(default="sync", description="'sync' (inline) or 'async' (job + poll)"),
+    debug: bool = Query(default=False),
+) -> dict[str, Any]:
+    """Parse a Legal Aid monthly payment-report PDF via Claude Vision.
+
+    Auth: X-Api-Key.
+    Body: multipart with `file` (the PDF bytes).
+    Optional `subject` query: if supplied, used to extract `[סימוכין NNN]` →
+        sysref. Manual uploads typically omit this.
+    Optional `mode=async`: returns `{jobId, status:"pending"}` immediately and
+        runs the parse in the background; poll `/parse-pdf-report/jobs/{jobId}`.
+        Default `sync` runs inline (the vision call can take several minutes —
+        only safe for callers without a short request timeout, e.g. n8n).
+    Optional `debug=true`: also returns the raw model response + page count.
+    """
+    _verify_admin(request)
+
+    pdf_bytes = await file.read()
+    # Validate up front so an obviously-bad upload fails fast in BOTH modes
+    # (rather than surfacing as a background-job error a moment later).
+    if not pdf_bytes:
+        raise HTTPException(status_code=400, detail="Empty file")
+    if pdf_bytes[:4] != b"%PDF":
+        raise HTTPException(status_code=400, detail="File is not a PDF")
+
+    if mode == "async":
+        _cleanup_old_jobs()
+        job_id = uuid.uuid4().hex
+        _write_job(job_id, {"jobId": job_id, "status": "pending", "createdAt": time.time()})
+        task = asyncio.create_task(_run_parse_job(job_id, pdf_bytes, subject, debug))
+        _running_jobs.add(task)  # keep a strong ref so the task isn't GC'd mid-run
+        task.add_done_callback(_running_jobs.discard)
+        return {"jobId": job_id, "status": "pending"}
+
+    return await _parse_pdf_to_payload(pdf_bytes, subject, debug)
+
+
+@router.get("/parse-pdf-report/jobs/{job_id}")
+async def get_parse_job(request: Request, job_id: str) -> dict[str, Any]:
+    """Poll an async parse job. Returns `{status: pending|done|error, ...}`.
+
+    `done` carries `result` (the ingest payload); `error` carries `detail`.
+    """
+    _verify_admin(request)
+    job = _read_job(job_id)
+    if job is None:
+        raise HTTPException(status_code=404, detail="Job not found (expired or invalid)")
+    return job
+
+
+# ----------------------------------------------------------------------
+# AI Gateway / Claude Vision call
+# ----------------------------------------------------------------------
+
+
+async def _call_vision_for_json(
+    images_b64: list[tuple[str, str]], prompt: str
+) -> str:
+    """Send page images to Claude Vision via ai-gateway and return raw text.
+
+    Mirrors the pattern in api/services/ocr.py — uses the OpenAI-compatible
+    chat completions endpoint that ai-gateway exposes at /v1.
+    """
+    try:
+        from openai import AsyncOpenAI
+    except ImportError as e:
+        raise HTTPException(status_code=503, detail=f"openai sdk missing: {e}")
+
+    base_url = os.environ.get("AI_GATEWAY_URL", "http://localhost:3000").rstrip("/") + "/v1"
+    api_key = os.environ.get("AI_GATEWAY_API_KEY", "")
+    if not api_key:
+        raise HTTPException(
+            status_code=503,
+            detail="AI_GATEWAY_API_KEY is not configured on shira-hermes",
+        )
+
+    client = AsyncOpenAI(base_url=base_url, api_key=api_key)
+
+    content: list[dict[str, Any]] = [{"type": "text", "text": prompt}]
+    for mime, b64 in images_b64:
+        content.append(
+            {
+                "type": "image_url",
+                "image_url": {"url": f"data:{mime};base64,{b64}"},
+            }
+        )
+
+    try:
+        response = await client.chat.completions.create(
+            model=os.environ.get("CLAUDE_MODEL", "sonnet"),
+            messages=[{"role": "user", "content": content}],
+            max_tokens=4096,
+            temperature=0,
+        )
+    except Exception as e:
+        logger.exception("Legal Aid: ai-gateway call failed")
+        raise HTTPException(status_code=502, detail=f"ai-gateway error: {e}")
+
+    return (response.choices[0].message.content or "").strip()
+
+
+# ----------------------------------------------------------------------
+# Response parsing + normalization
+# ----------------------------------------------------------------------
+
+
+def _parse_json_response(text: str) -> dict[str, Any]:
+    """Extract a JSON object from the model's raw text.
+
+    Handles a few common Claude failure modes:
+      - markdown fences (```json ... ```)
+      - trailing commentary after the JSON block
+      - smart-quotes in keys/values
+    """
+    if not text:
+        return {}
+
+    # Strip code fences first.
+    fence = re.match(r"^\s*```(?:json)?\s*\n(.*?)\n?\s*```\s*$", text, re.DOTALL)
+    if fence:
+        text = fence.group(1)
+
+    # Take the substring from first `{` to matching last `}`.
+    start = text.find("{")
+    end = text.rfind("}")
+    if start == -1 or end == -1 or end <= start:
+        logger.warning("Legal Aid: no JSON object found in model response: %r", text[:300])
+        return {}
+
+    candidate = text[start : end + 1]
+
+    try:
+        return json.loads(candidate)
+    except json.JSONDecodeError:
+        # Try one fix-up pass: replace smart quotes, then retry.
+        cleaned = (
+            candidate.replace("“", '"')
+            .replace("”", '"')
+            .replace("‘", "'")
+            .replace("’", "'")
+        )
+        try:
+            return json.loads(cleaned)
+        except json.JSONDecodeError as e:
+            logger.warning(
+                "Legal Aid: model response is not valid JSON: %s; raw=%r",
+                e,
+                candidate[:300],
+            )
+            return {}
+
+
+def _normalize_lines(lines: list[Any]) -> list[dict[str, Any]]:
+    out: list[dict[str, Any]] = []
+    for raw in lines:
+        if not isinstance(raw, dict):
+            continue
+        # Drop rows that don't carry the minimum identifying info.
+        if not (raw.get("invoiceNumber") or raw.get("legalAidCaseNumber")):
+            continue
+        out.append(
+            {
+                "invoiceNumber": _str_or_none(raw.get("invoiceNumber")),
+                "requestNumber": _str_or_none(raw.get("requestNumber")),
+                "legalAidCaseNumber": _str_or_none(raw.get("legalAidCaseNumber")),
+                "clientName": _str_or_none(raw.get("clientName")),
+                "activityCategory": _str_or_none(raw.get("activityCategory")),
+                "activitySubTypeHebrew": _str_or_none(raw.get("activitySubTypeHebrew")),
+                "activityDate": _str_or_none(raw.get("activityDate")),
+                "amountExclVat": _to_float(raw.get("amountExclVat")),
+                "vat": _to_float(raw.get("vat")),
+                "amountInclVat": _to_float(raw.get("amountInclVat")),
+            }
+        )
+    return out
+
+
+# ----------------------------------------------------------------------
+# Helpers
+# ----------------------------------------------------------------------
+
+
+def _to_float(v: Any) -> float:
+    if v is None:
+        return 0.0
+    if isinstance(v, (int, float)):
+        return float(v)
+    s = str(v).replace(",", "").replace("₪", "").strip()
+    try:
+        return float(s)
+    except ValueError:
+        return 0.0
+
+
+def _str_or_none(v: Any) -> str | None:
+    if v is None:
+        return None
+    s = str(v).strip()
+    return s or None

api/services/agent_runner.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import json
 import logging
 from typing import Any, Callable
@@ -14,6 +15,8 @@ from mcp_server.tools.document_tools import register_document_tools
 from mcp_server.tools.legal_tools import register_legal_tools
 from mcp_server.tools.legal_kb_tools import register_legal_kb_tools
 from mcp_server.tools.signature_tools import register_signature_tools
+from mcp_server.tools.billing_tools import register_billing_tools
+from mcp_server.tools.legal_aid_tools import register_legal_aid_tools
 from api.services.skills import register_skill_tools
 from api.services.memory import register_memory_tools
 from api.services.delegate import register_delegate_tool
@@ -82,6 +85,7 @@ class AgentRunner:
         model: str = "sonnet",
         max_tokens: int = 4096,
         max_iterations: int = 10,
+        total_timeout: float = 570.0,
     ):
         self.client = AsyncOpenAI(
             base_url=f"{ai_gateway_url.rstrip('/')}/v1",
@@ -90,6 +94,7 @@ class AgentRunner:
         self.model = model
         self.max_tokens = max_tokens
         self.max_iterations = max_iterations
+        self.total_timeout = total_timeout
 
     async def run(
         self,
@@ -106,12 +111,50 @@ class AgentRunner:
         kb_topic_name: str | None = None,
         on_event: Callable[[dict], None] | None = None,
     ) -> str:
-        """Run a conversation with tool-calling loop. Returns the final text response.
+        """Run a conversation with tool-calling loop, bounded by total_timeout seconds."""
+        try:
+            return await asyncio.wait_for(
+                self._run_impl(
+                    system_prompt=system_prompt,
+                    messages=messages,
+                    context=context,
+                    espocrm_url=espocrm_url,
+                    espocrm_api_key=espocrm_api_key,
+                    depth=depth,
+                    blocked_tools=blocked_tools,
+                    allowed_toolsets=allowed_toolsets,
+                    kb_sources_used=kb_sources_used,
+                    kb_topic_id=kb_topic_id,
+                    kb_topic_name=kb_topic_name,
+                    on_event=on_event,
+                ),
+                timeout=self.total_timeout,
+            )
+        except asyncio.TimeoutError:
+            logger.warning("[agent] total timeout (%.0fs) reached", self.total_timeout)
+            return "⏱️ המשימה לקחה יותר זמן מהמותר. נסי לפצל אותה לחלקים קטנים — למשל: קודם בקשי לקרוא מסמך ספציפי, ואז לכתוב את הסיכום."
+
+    async def _run_impl(
+        self,
+        system_prompt: str,
+        messages: list[dict[str, str]],
+        context: dict,
+        espocrm_url: str,
+        espocrm_api_key: str,
+        depth: int = 0,
+        blocked_tools: set[str] | None = None,
+        allowed_toolsets: list[str] | None = None,
+        kb_sources_used: list[dict] | None = None,
+        kb_topic_id: int | None = None,
+        kb_topic_name: str | None = None,
+        on_event: Callable[[dict], None] | None = None,
+    ) -> str:
+        """Internal tool-calling loop. Returns the final text response.
 
         Args:
             depth: Current delegation depth. 0 = main conversation, 1 = child agent.
             blocked_tools: Tool names to exclude (used by delegation to prevent recursion).
-            allowed_toolsets: If set, only register tools from these categories (crm, documents, legal).
+            allowed_toolsets: If set, only register tools from these categories (crm, documents, legal, signature, billing, legal_aid).
             on_event: Optional sync callback fired between iterations and tool calls.
                 Receives a dict {type, label, ...} where type ∈ {thinking,
                 tool_start, tool_done, tool_error}. Used by /kb/ask/stream
@@ -150,6 +193,10 @@ class AgentRunner:
             )
         if should_register_all or "signature" in (allowed_toolsets or []):
             register_signature_tools(tools_registry, crm, case_id, user_id, context)
+        if should_register_all or "billing" in (allowed_toolsets or []):
+            register_billing_tools(tools_registry, crm, case_id, user_id, context)
+        if should_register_all or "legal_aid" in (allowed_toolsets or []):
+            register_legal_aid_tools(tools_registry, crm, case_id, user_id, context)
 
         # Phase 3 tools (only for main conversation, not child agents)
         if depth == 0:

api/services/kb/admin_sources.py

@@ -4,22 +4,31 @@ Today the only way to clean up or correct a source is `DELETE FROM
 kb_source WHERE id=...` on the shared PG plus an mc rm in MinIO. This
 module backs the new sources table in the EspoCRM "ניהול" tab:
 
-  list_sources       — table view with chunk count and last-ingest status
-  update_source      — patch the user-editable metadata
-                       (title, identifier, dates, source_url)
-  delete_source      — hard delete (cascades chunks; drops S3 object)
-  start_reingest     — re-parse the existing file in place; preserves
-                       source_id and metadata, replaces chunks
-  process_reingest_job — async background worker that does the actual
-                         parse→chunk→embed and swaps chunks in a single
-                         transaction (UI polls the same /admin/kb/jobs
-                         endpoint as upload jobs)
+  list_sources              — table view with chunk count and last-ingest status
+  update_source             — patch the user-editable metadata
+                              (title, identifier, dates, source_url)
+  delete_source             — hard delete (cascades chunks; drops S3 object)
+  start_reingest            — kick off the two-phase re-ingest pipeline:
+                              classify → awaiting_review → commit. The
+                              classify stage reuses
+                              ingest_jobs.process_classify_stage so the
+                              user gets fresh AI metadata suggestions in
+                              the same review UI as fresh uploads.
+  process_reingest_embed_stage — final stage after the user commits:
+                              UPDATE kb_source metadata + replace labels +
+                              re-chunk and re-embed (the original
+                              process_reingest_job logic, minus the
+                              "preserve metadata" rule).
 
 The reingest path differs from ingest_source: ingest_source creates a
 NEW kb_source row and supersedes the previous version. Reingest keeps
 the same row id (so cached _lastSearch chunk_index references in the
 browser still point at a valid source), wipes its chunks, and inserts
-fresh ones from the same file. Hand-edited metadata is preserved.
+fresh ones from the same file. As of v0.9.0, hand-edited metadata is
+NO LONGER preserved silently — the user reviews AI suggestions and
+edits them on the batch review screen before committing, so any
+overwrite is explicit. Sources whose AI-extracted metadata was thin
+(e.g. title="החלטה" only) get fresh suggestions on every re-process.
 """
 
 from __future__ import annotations
@@ -244,14 +253,22 @@ async def delete_source(source_id: int) -> dict:
 
 # ── Re-ingest ───────────────────────────────────────────────────────────────
 
-async def start_reingest(source_id: int, requested_by_user: Optional[str]) -> int:
+async def start_reingest(
+    source_id: int, requested_by_user: Optional[str]
+) -> dict:
     """Create a kb_ingest_job tied to the existing source_id.
 
-    The actual work is in `process_reingest_job`, scheduled via
-    `asyncio.create_task` from the route. Returns the new job_id so the
-    browser can poll using the same /admin/kb/jobs/{id} endpoint as
-    upload jobs.
+    The job enters the same two-phase pipeline as a fresh upload —
+    classify (AI metadata extraction) → awaiting_review → embed
+    (re-chunk + UPDATE kb_source) — so the user can review suggestions
+    on the batch review screen before anything overwrites kb_source.
+
+    Returns {"job_id", "batch_id"}; the route schedules
+    `ingest_jobs.process_classify_stage(job_id)` after this, and the
+    browser opens `/admin/kb/batch/<batch_id>` to poll the review UI.
     """
+    import uuid as _uuid
+
     src = await get_source(source_id)
     if not src:
         raise LookupError(f"source {source_id} not found")
@@ -271,58 +288,88 @@ async def start_reingest(source_id: int, requested_by_user: Optional[str]) -> in
         raise ValueError(f"original file unavailable: {exc}")
 
     # Reuse the kb_ingest_job table; mark this row as a re-ingest by
-    # pre-filling source_id (uploads start with source_id NULL).
+    # pre-filling source_id (uploads start with source_id NULL). The
+    # batch_id is auto-generated so the existing review UI works
+    # without changes — re-ingests are single-file batches. Existing
+    # labels go into metadata_json so the review form can pre-fill
+    # them as a fallback when the AI classifier hasn't run yet (or
+    # returns empty).
+    from api.services.kb import labels as kb_labels  # local import: cycle
+    existing_labels = await kb_labels.get_labels_for_source(source_id)
     metadata = {
         "title": src.get("title"),
         "identifier": src.get("identifier"),
         "published_at": _date_to_iso(src.get("published_at")),
         "effective_at": _date_to_iso(src.get("effective_at")),
         "source_url": src.get("source_url"),
+        "summary": src.get("description"),
+        "labels": [lbl.get("name") for lbl in existing_labels if lbl.get("name")],
         "reingest": True,
     }
+    batch_id = _uuid.uuid4().hex
+    # Use the actual filename (with extension) — NOT the source title.
+    # parse_first_pages dispatches by extension (.pdf/.docx/.txt); a
+    # title like "החלטה" without an extension makes the parser bail
+    # and return empty text, which then makes the classifier return {}.
+    # See incident 2026-04-28: source 141 (title="החלטה") had every
+    # re-ingest classify as empty until this was fixed.
     pool = await get_pool()
     async with pool.acquire() as conn:
         row = await conn.fetchrow(
             """
             INSERT INTO kb_ingest_job
                 (source_id, original_filename, s3_key, kind, topic_id,
-                 metadata_json, status, requested_by_user)
-            VALUES ($1, $2, $3, $4, $5, $6::jsonb, 'queued', $7)
+                 metadata_json, status, requested_by_user, batch_id)
+            VALUES ($1, $2, $3, $4, $5, $6::jsonb, 'queued', $7, $8)
             RETURNING id
             """,
             source_id,
-            src.get("title") or _filename_from_key(key),
+            _filename_from_key(key),
             key,
             src["kind"],
             src["topic_id"],
             json.dumps(metadata, ensure_ascii=False),
             requested_by_user,
+            batch_id,
         )
     job_id = row["id"]
-    logger.info("[admin.reingest] queued job_id=%s source_id=%s", job_id, source_id)
-    return job_id
+    logger.info(
+        "[admin.reingest] queued job_id=%s source_id=%s batch_id=%s",
+        job_id, source_id, batch_id,
+    )
+    return {"job_id": job_id, "batch_id": batch_id}
 
 
-async def process_reingest_job(job_id: int) -> None:
-    """Background worker for re-ingest. Replaces chunks in-place; preserves
-    the kb_source row and its metadata.
+async def process_reingest_embed_stage(
+    job_id: int, fields: dict, label_ids: list[int],
+) -> None:
+    """Final stage of the two-phase re-ingest pipeline.
 
-    Errors are written to the job row (status='failed', error_message=...)
-    so the polling UI never gets stuck on 'processing'.
+    Called from `ingest_jobs.commit_job` after the user reviewed the AI
+    suggestions. The job is already in `status='processing',
+    processing_stage='embedding'` at this point — commit_job set that
+    transition before scheduling us.
+
+    Steps:
+      1. UPDATE kb_source with the user-confirmed metadata (kind,
+         title, identifier, source_url, dates, description=summary).
+      2. Replace the source's labels with `label_ids`.
+      3. Re-parse the file → chunk → embed → atomic DELETE+INSERT
+         of kb_chunk rows + checksum update.
+      4. Mark the job done.
+
+    Failures here are real failures (the user already confirmed). They
+    surface as `status='failed'` so the polling UI lets the user retry.
     """
+    from api.services.kb import labels as kb_labels  # local import: cycle
+
     pool = await get_pool()
     async with pool.acquire() as conn:
         job_row = await conn.fetchrow(
-            """
-            UPDATE kb_ingest_job
-               SET status = 'processing', started_at = now()
-             WHERE id = $1 AND status = 'queued'
-            RETURNING *
-            """,
-            job_id,
+            "SELECT * FROM kb_ingest_job WHERE id = $1", job_id,
         )
     if not job_row:
-        logger.warning("[reingest] job %s not in queued state, skipping", job_id)
+        logger.warning("[reingest-embed] job %s vanished", job_id)
         return
 
     job = _row_to_dict(job_row)
@@ -332,11 +379,34 @@ async def process_reingest_job(job_id: int) -> None:
         return
 
     try:
-        bucket, key = _parse_s3_uri(job["s3_key"])  # s3_key here is just the key
-        # If s3_key is a bare key (no s3:// prefix), _parse_s3_uri returns
-        # ("", "") — fall back to the bare key.
+        # Step 1: UPDATE kb_source with the user-confirmed metadata.
+        # commit_job's `fields` dict mirrors JobCommit: kind, title,
+        # identifier, source_url, published_at, effective_at, summary.
+        # kb_source.description is the schema name for "summary".
+        kind = (fields.get("kind") or job["kind"]).strip()
+        title = (fields.get("title") or "").strip()
+        if not title:
+            raise ValueError("title is empty after commit — aborting re-ingest")
+
+        await _update_source_metadata_for_reingest(
+            pool=pool,
+            source_id=source_id,
+            kind=kind,
+            title=title,
+            identifier=fields.get("identifier"),
+            source_url=fields.get("source_url"),
+            published_at=fields.get("published_at"),
+            effective_at=fields.get("effective_at"),
+            description=fields.get("summary"),
+        )
+
+        # Step 2: replace labels (empty list clears all).
+        await kb_labels.replace_for_source(source_id, label_ids or [])
+
+        # Step 3: re-chunk + re-embed.
+        bucket, key = _parse_s3_uri(job["s3_key"])
         if not key:
-            key = job["s3_key"]
+            key = job["s3_key"]  # bare key, no s3:// prefix
         data = kb_s3._client().get_object(
             Bucket=bucket or kb_s3._bucket(),
             Key=key,
@@ -348,7 +418,7 @@ async def process_reingest_job(job_id: int) -> None:
         if not text:
             raise kb_ingest.IngestError("no text extracted")
 
-        chunks = chunk_text(text, job["kind"])
+        chunks = chunk_text(text, kind)
         if not chunks:
             raise kb_ingest.IngestError("chunker produced zero chunks")
 
@@ -388,8 +458,6 @@ async def process_reingest_job(job_id: int) -> None:
                     """,
                     rows,
                 )
-                # Update only the structural fields that depend on the file
-                # (checksum). Hand-edited metadata stays as-is.
                 await conn.execute(
                     "UPDATE kb_source SET checksum = $2 WHERE id = $1",
                     source_id, new_checksum,
@@ -398,6 +466,7 @@ async def process_reingest_job(job_id: int) -> None:
                     """
                     UPDATE kb_ingest_job
                        SET status = 'done',
+                           processing_stage = NULL,
                            chunks_created = $2,
                            completed_at = now()
                      WHERE id = $1
@@ -405,14 +474,50 @@ async def process_reingest_job(job_id: int) -> None:
                     job_id, len(chunks),
                 )
         logger.info(
-            "[reingest] done job_id=%s source_id=%s chunks=%d",
-            job_id, source_id, len(chunks),
+            "[reingest-embed] done job_id=%s source_id=%s chunks=%d labels=%d",
+            job_id, source_id, len(chunks), len(label_ids or []),
         )
     except Exception as exc:
-        logger.exception("[reingest] failed job_id=%s", job_id)
+        logger.exception("[reingest-embed] failed job_id=%s", job_id)
         await _fail_job(pool, job_id, str(exc))
 
 
+async def _update_source_metadata_for_reingest(
+    *, pool, source_id: int, kind: str, title: str,
+    identifier: Optional[str], source_url: Optional[str],
+    published_at: Optional[str], effective_at: Optional[str],
+    description: Optional[str],
+) -> None:
+    """UPDATE kb_source after a re-ingest commit. Bypasses _EDITABLE_FIELDS
+    so we can also update `kind` (the user explicitly chose it on the
+    review screen — e.g. caselaw uploaded as circular by mistake).
+    """
+    pub = kb_ingest._coerce_date(published_at) if published_at else None
+    eff = kb_ingest._coerce_date(effective_at) if effective_at else None
+    async with pool.acquire() as conn:
+        await conn.execute(
+            """
+            UPDATE kb_source
+               SET kind = $2,
+                   title = $3,
+                   identifier = $4,
+                   source_url = $5,
+                   published_at = $6,
+                   effective_at = $7,
+                   description = $8
+             WHERE id = $1
+            """,
+            source_id,
+            kind,
+            title,
+            (identifier or None),
+            (source_url or None),
+            pub,
+            eff,
+            (description or None),
+        )
+
+
 # ── helpers ─────────────────────────────────────────────────────────────────
 
 async def _fail_job(pool, job_id: int, msg: str) -> None:

api/services/kb/classifier.py

@@ -42,10 +42,13 @@ logger = logging.getLogger("shira.kb.classifier")
 # it's no worse than 1.
 _CONCURRENCY = int(os.environ.get("KB_CLASSIFIER_CONCURRENCY", "2"))
 # Empirically the ai-gateway → Claude tool-call round-trip with a ~3KB
-# Hebrew system prompt + 5KB user content runs 18-25s. With 20 files
-# queued through the gateway one-at-a-time, the tail can hit ~80s.
-# Padded to 90s so we don't fail-soft on serialized queue waits.
-_TIMEOUT_S = float(os.environ.get("KB_CLASSIFIER_TIMEOUT_S", "90"))
+# Hebrew system prompt + 5KB user content runs 18-25s. Long Hebrew
+# caselaw PDFs (full 20K-char first-3-pages cap) push the tail to
+# ~92s on DOCX and noticeably more on dense PDFs; ai-gateway also
+# serializes through one OAuth session, lengthening the tail in a
+# batch. 240s buys headroom without making the user wait forever
+# when something is genuinely stuck.
+_TIMEOUT_S = float(os.environ.get("KB_CLASSIFIER_TIMEOUT_S", "240"))
 _MAX_INPUT_CHARS = 20_000   # ~3 pages of Hebrew legal text
 
 _semaphore = asyncio.Semaphore(_CONCURRENCY)

api/services/kb/ingest_jobs.py

@@ -389,6 +389,11 @@ async def commit_job(
     `label_specs` is a list of {slug, name, topic_id?} — existing labels
     by slug, new ones by display name. lookup_or_create_batch resolves
     them all to ids race-safely.
+
+    Branches on `job.source_id`:
+      • NULL  → fresh upload → `process_embed_stage` (creates kb_source).
+      • set   → re-ingest    → `process_reingest_embed_stage` (UPDATEs
+                              the existing kb_source + replaces chunks).
     """
     pool = await get_pool()
 
@@ -426,7 +431,17 @@ async def commit_job(
             json.dumps(fields, ensure_ascii=False),
         )
 
-    asyncio.create_task(process_embed_stage(job_id, fields, label_ids))
+    if job.get("source_id"):
+        # Re-ingest: UPDATE the existing kb_source + replace chunks.
+        from api.services.kb import admin_sources as kb_admin_sources
+        asyncio.create_task(
+            kb_admin_sources.process_reingest_embed_stage(
+                job_id, fields, label_ids,
+            )
+        )
+    else:
+        # Fresh upload: ingest_source creates a new kb_source row.
+        asyncio.create_task(process_embed_stage(job_id, fields, label_ids))
     return {"job_id": job_id, "status": "processing", "label_ids": label_ids}
 
 
@@ -522,11 +537,22 @@ async def discard_job(job_id: int, requested_by: str | None) -> dict:
     if job["status"] in ("done",):
         raise ValueError(f"job {job_id} already done — cannot discard")
 
-    # Best-effort S3 cleanup; the row stays as audit trail.
-    try:
-        kb_s3._client().delete_object(Bucket=kb_s3._bucket(), Key=job["s3_key"])
-    except Exception as e:
-        logger.warning("[discard] S3 delete failed for %s: %s", job["s3_key"], e)
+    # Best-effort S3 cleanup — but ONLY for fresh uploads (source_id IS
+    # NULL). For re-ingest jobs (source_id set), the s3_key points to
+    # the existing source's file in kb_source.original_path; deleting
+    # it would orphan the source and break any future re-process.
+    # See incident 2026-04-28: source 141 lost its PDF when a discarded
+    # re-ingest deleted the underlying file.
+    if job.get("source_id") is None:
+        try:
+            kb_s3._client().delete_object(Bucket=kb_s3._bucket(), Key=job["s3_key"])
+        except Exception as e:
+            logger.warning("[discard] S3 delete failed for %s: %s", job["s3_key"], e)
+    else:
+        logger.info(
+            "[discard] keeping S3 file for re-ingest job %s (source_id=%s)",
+            job_id, job["source_id"],
+        )
 
     reason = f"discarded by {requested_by or 'user'}"
     async with pool.acquire() as conn:

api/services/kb/s3.py

@@ -1,4 +1,4 @@
-"""MinIO/S3 helpers for the insurance-kb bucket.
+"""MinIO/S3 helpers for the shared kb bucket (formerly `insurance-kb`).
 
 Layout in the bucket:
     inbox/{law,regulation,circular,caselaw,tool}/<file>
@@ -41,7 +41,7 @@ def _client():
 
 
 def _bucket() -> str:
-    return os.environ.get("KB_S3_BUCKET", "insurance-kb")
+    return os.environ.get("KB_S3_BUCKET", "legal-kb")
 
 
 def list_inbox() -> list[dict]:

api/services/kb/voyage.py

@@ -2,8 +2,10 @@
 
 from __future__ import annotations
 
+import asyncio
 import logging
 import os
+import random
 from typing import Literal
 
 import httpx
@@ -20,11 +22,58 @@ _MAX_BATCH_TOKENS = 90_000
 # Voyage's multilingual tokenizer splits Hebrew more aggressively than ASCII.
 _TOKENS_PER_CHAR_EST = 0.55
 
+# Retry parameters for transient failures (429 + 5xx). Dev and prod share a
+# single VOYAGE_API_KEY; without retry, simultaneous bulk re-ingests on both
+# environments hit the per-key RPM limit and fail one of them. With
+# exponential backoff (jittered) the burst is absorbed in seconds.
+_MAX_RETRIES = 5
+_BASE_BACKOFF_S = 2.0
+_MAX_BACKOFF_S = 30.0
+
 
 class VoyageError(RuntimeError):
     pass
 
 
+def _retry_after_seconds(resp: httpx.Response, attempt: int) -> float:
+    """Honor `Retry-After` header if present, else exponential backoff."""
+    hdr = resp.headers.get("retry-after")
+    if hdr:
+        try:
+            return min(_MAX_BACKOFF_S, max(0.1, float(hdr)))
+        except ValueError:
+            pass
+    backoff = min(_MAX_BACKOFF_S, _BASE_BACKOFF_S * (2 ** attempt))
+    return backoff + random.uniform(0, backoff * 0.25)
+
+
+async def _post_with_retry(
+    http: httpx.AsyncClient, url: str, *, headers: dict, json: dict, what: str,
+) -> httpx.Response:
+    """POST with exponential backoff on 429 and 5xx.
+
+    Returns the final Response (caller still checks status_code for
+    non-retryable errors, e.g. 4xx other than 429).
+    """
+    last_resp: httpx.Response | None = None
+    for attempt in range(_MAX_RETRIES + 1):
+        resp = await http.post(url, headers=headers, json=json)
+        last_resp = resp
+        if resp.status_code == 200:
+            return resp
+        if resp.status_code != 429 and resp.status_code < 500:
+            return resp
+        if attempt == _MAX_RETRIES:
+            return resp
+        delay = _retry_after_seconds(resp, attempt)
+        logger.warning(
+            "[kb.voyage] %s got %d, retry %d/%d in %.1fs",
+            what, resp.status_code, attempt + 1, _MAX_RETRIES, delay,
+        )
+        await asyncio.sleep(delay)
+    return last_resp  # type: ignore[return-value]
+
+
 async def embed(
     texts: list[str],
     input_type: Literal["document", "query"] = "document",
@@ -56,10 +105,12 @@ async def embed(
     out: list[list[float]] = []
     async with httpx.AsyncClient(timeout=120) as http:
         for bi, batch in enumerate(batches):
-            resp = await http.post(
+            resp = await _post_with_retry(
+                http,
                 _VOYAGE_URL,
                 headers={"Authorization": f"Bearer {api_key}"},
                 json={"model": model, "input": batch, "input_type": input_type},
+                what=f"embed batch {bi+1}/{len(batches)}",
             )
             if resp.status_code != 200:
                 raise VoyageError(f"Voyage API {resp.status_code}: {resp.text[:200]}")
@@ -104,10 +155,12 @@ async def rerank(
         payload["top_k"] = int(top_k)
 
     async with httpx.AsyncClient(timeout=60) as http:
-        resp = await http.post(
+        resp = await _post_with_retry(
+            http,
             _VOYAGE_RERANK_URL,
             headers={"Authorization": f"Bearer {api_key}"},
             json=payload,
+            what="rerank",
         )
     if resp.status_code != 200:
         raise VoyageError(f"Voyage rerank {resp.status_code}: {resp.text[:200]}")

api/services/memory.py

@@ -1,4 +1,12 @@
-"""Per-lawyer and per-case memory store — bounded markdown files with atomic writes."""
+"""Per-lawyer and per-case memory store — bounded markdown files with atomic writes.
+
+The filesystem files (under /opt/data/) remain the canonical write target for
+update_case_memory and update_user_profile. After each successful filesystem
+write, update_user_profile also syncs the result up to the EspoCRM UserProfile
+entity (write-through) so an admin browsing UserProfile in the UI sees the
+same content that's in the prompt. The CRM is the read source for the prompt
+(context['userProfile'] populated by UserProfileContextProvider).
+"""
 
 from __future__ import annotations
 
@@ -9,6 +17,8 @@ import re
 import tempfile
 from pathlib import Path
 
+import httpx
+
 logger = logging.getLogger("shira.memory")
 
 DATA_DIR = os.environ.get("SHIRA_DATA_DIR", "/opt/data")
@@ -79,6 +89,46 @@ def _truncate(content: str, max_chars: int) -> str:
     return _join_entries(entries)
 
 
+# ---------- CRM write-through ----------
+
+async def _sync_user_profile_to_crm(user_id: str, content: str, default_name: str | None = None) -> None:
+    """Write-through to UserProfile entity in EspoCRM. Best-effort; logs and swallows errors."""
+    base = (os.environ.get("ESPOCRM_URL") or "").rstrip("/")
+    key = os.environ.get("ESPOCRM_API_KEY") or ""
+    if not base or not key:
+        logger.debug("ESPOCRM_URL or API key missing — skipping UserProfile sync")
+        return
+
+    name = (default_name or f"Profile {user_id[:8]}")[:255]
+    headers = {"X-Api-Key": key, "Content-Type": "application/json"}
+    try:
+        async with httpx.AsyncClient(timeout=15.0) as client:
+            # Find existing row for this user
+            r = await client.get(
+                f"{base}/api/v1/UserProfile",
+                headers=headers,
+                params={
+                    "where[0][type]": "equals",
+                    "where[0][attribute]": "userId",
+                    "where[0][value]": user_id,
+                    "maxSize": 1,
+                },
+            )
+            existing = (r.json().get("list", []) if r.status_code == 200 else [])
+            row = existing[0] if existing else None
+
+            payload = {"content": content, "isActive": True}
+            if row:
+                resp = await client.put(f"{base}/api/v1/UserProfile/{row['id']}", headers=headers, json=payload)
+            else:
+                payload.update({"userId": user_id, "name": name})
+                resp = await client.post(f"{base}/api/v1/UserProfile", headers=headers, json=payload)
+            if resp.status_code >= 400:
+                logger.warning("UserProfile sync HTTP %s: %s", resp.status_code, resp.text[:200])
+    except Exception as e:
+        logger.warning("UserProfile sync failed: %s", e)
+
+
 # ---------- Public API ----------
 
 def load_user_profile(user_id: str) -> str:
@@ -102,6 +152,7 @@ async def update_user_profile(user_id: str, action: str, content: str, match: st
             entries.append(content)
             result = _truncate(_join_entries(entries), PROFILE_MAX_CHARS)
             _write_file_atomic(path, result)
+            await _sync_user_profile_to_crm(user_id, result)
             return f'✅ נוסף לפרופיל: "{content[:60]}"'
 
         elif action == "replace":
@@ -117,6 +168,7 @@ async def update_user_profile(user_id: str, action: str, content: str, match: st
                 return f'❌ לא נמצאה רשומה עם "{match}"'
             result = _truncate(_join_entries(entries), PROFILE_MAX_CHARS)
             _write_file_atomic(path, result)
+            await _sync_user_profile_to_crm(user_id, result)
             return f'✅ רשומה עודכנה בפרופיל.'
 
         elif action == "remove":
@@ -126,7 +178,9 @@ async def update_user_profile(user_id: str, action: str, content: str, match: st
             entries = [e for e in entries if match not in e]
             if len(entries) == original_len:
                 return f'❌ לא נמצאה רשומה עם "{match}"'
-            _write_file_atomic(path, _join_entries(entries))
+            result = _join_entries(entries)
+            _write_file_atomic(path, result)
+            await _sync_user_profile_to_crm(user_id, result)
             return "✅ רשומה הוסרה מהפרופיל."
 
         elif action == "read":

api/services/ocr.py

@@ -84,32 +84,75 @@ async def ocr_pdf(base64_data: str) -> str:
     return "\n\n".join(pages_text)
 
 
+def _extract_docx_xml_text(docx_zip) -> str:
+    """Parse word/document.xml from a DOCX and return plain text.
+
+    The DOCX text payload is a flat sequence of <w:t>...</w:t> runs grouped
+    inside <w:p> paragraphs. We turn each paragraph into one text line.
+    """
+    import re
+    try:
+        xml = docx_zip.read("word/document.xml").decode("utf-8", errors="replace")
+    except KeyError:
+        return ""
+
+    # Anchor the run-open to either `<w:t>` (no attrs) or `<w:t ...>` (with attrs).
+    # The `(?:\s[^>]*)?` keeps us from also matching `<w:tab>` / `<w:tbl>` etc.
+    run_re = re.compile(r"<w:t(?:\s[^>]*)?>(.*?)</w:t>", flags=re.DOTALL)
+    para_re = re.compile(r"<w:p(?:\s[^>]*)?>.*?</w:p>", flags=re.DOTALL)
+
+    paragraphs = []
+    for p_match in para_re.finditer(xml):
+        block = p_match.group(0)
+        line = "".join(run_re.findall(block)).strip()
+        if line:
+            paragraphs.append(line)
+    return "\n".join(paragraphs).strip()
+
+
 async def ocr_docx_images(base64_data: str) -> str:
-    """Extract images from a DOCX archive and OCR each via Claude Vision."""
+    """Extract text + images from a DOCX archive.
+
+    Order of operations:
+    1. Read word/document.xml for the actual text content (the LLM needs this).
+    2. OCR any embedded images via Claude Vision (signatures, scanned inserts).
+
+    Previously this only OCR'd images, so a text-heavy DOCX with one signature
+    image returned only `[signature]` — and downstream LLMs hallucinated the rest.
+    """
     import zipfile
 
     docx_bytes = base64.b64decode(base64_data)
-    image_extracts = []
     try:
-        with zipfile.ZipFile(io.BytesIO(docx_bytes)) as z:
-            media = [n for n in z.namelist() if n.startswith("word/media/")]
-            logger.info("[ocr] docx images found=%d", len(media))
-            for i, name in enumerate(media[:MAX_PDF_PAGES]):
-                ext = name.rsplit(".", 1)[-1].lower()
-                mime = {"png": "image/png", "jpg": "image/jpeg", "jpeg": "image/jpeg", "gif": "image/gif"}.get(ext, "image/png")
-                img_bytes = z.read(name)
-                img_b64 = base64.b64encode(img_bytes).decode("ascii")
-                try:
-                    text = await _call_vision([(mime, img_b64)])
-                    image_extracts.append(f"=== תמונה {i + 1} ({name}) ===\n{text}")
-                except Exception as e:
-                    logger.error("[ocr] docx image %s failed: %s", name, e)
+        z = zipfile.ZipFile(io.BytesIO(docx_bytes))
     except zipfile.BadZipFile:
         return "❌ הקובץ אינו DOCX תקין."
 
-    if not image_extracts:
-        return "❌ לא נמצאו תמונות ב-DOCX לחילוץ טקסט."
-    return "\n\n".join(image_extracts)
+    sections = []
+
+    # 1. Text from document.xml
+    xml_text = _extract_docx_xml_text(z)
+    logger.info("[ocr] docx xml text chars=%d", len(xml_text))
+    if xml_text:
+        sections.append(f"=== טקסט מהמסמך ({len(xml_text)} תווים) ===\n{xml_text}")
+
+    # 2. OCR each embedded image
+    media = [n for n in z.namelist() if n.startswith("word/media/")]
+    logger.info("[ocr] docx images found=%d", len(media))
+    for i, name in enumerate(media[:MAX_PDF_PAGES]):
+        ext = name.rsplit(".", 1)[-1].lower()
+        mime = {"png": "image/png", "jpg": "image/jpeg", "jpeg": "image/jpeg", "gif": "image/gif"}.get(ext, "image/png")
+        img_bytes = z.read(name)
+        img_b64 = base64.b64encode(img_bytes).decode("ascii")
+        try:
+            text = await _call_vision([(mime, img_b64)])
+            sections.append(f"=== תמונה {i + 1} ({name}) ===\n{text}")
+        except Exception as e:
+            logger.error("[ocr] docx image %s failed: %s", name, e)
+
+    if not sections:
+        return "❌ לא נמצא טקסט וגם לא תמונות ב-DOCX."
+    return "\n\n".join(sections)
 
 
 async def extract_text_from_bytes(base64_data: str, mime_type: str, file_name: str = "") -> str:

api/services/prompt_builder.py

@@ -37,13 +37,36 @@ def _to_local(dt_str: str | None, fallback: str = "") -> str:
         return dt_str
 
 
-TOOL_RULES = (
+def _get_prompt(context: dict, key: str, default: str) -> str:
+    """Pull a prompt section override from context['assistantPrompts'], or fall back to the in-code default.
+
+    Empty strings and missing keys both fall back. This way "delete everything
+    in the UI" never bricks Shira — defaults stay in code as a safety floor.
+    """
+    overrides = (context.get("assistantPrompts") or {})
+    value = overrides.get(key)
+    return value.strip() if isinstance(value, str) and value.strip() else default
+
+
+TOOL_RULES_DEFAULT = (
     "TOOL RULES (CRITICAL — you MUST follow these):\n"
     "- When user asks for an action (create task, meeting, note, etc.) — USE the tool. Do not describe what you would do — DO IT.\n"
     "- NEVER claim you performed an action if you did not call the tool. This is the most important rule.\n"
     "- If a tool call fails, tell the user exactly what went wrong. Never pretend it succeeded.\n"
     "- If you cannot perform an action (missing context, wrong mode), explain why honestly.\n"
     "- After a successful tool call, you will receive the result. Only THEN confirm to the user what happened.\n"
+    "- VERIFY TOOL RESULT (absolute rule — violations cause real production damage):\n"
+    "  * Every tool result must be READ before you write your response. The first character tells you if it succeeded.\n"
+    "  * If the result starts with '❌' or contains the words 'שגיאה'/'נכשל'/'failed'/'error' — the action FAILED. Report the failure honestly to the user, do not say '✅ נשמר' / '✅ נוצר' / 'הכל תועד'.\n"
+    "  * HTTP errors (404, 500, etc.) returned from a tool also mean failure. Never paper over them.\n"
+    "  * Do not interpret your own previous statement as proof of success. The ONLY proof is the tool result.\n"
+    "- DOCUMENT FAITHFULNESS (absolute rule — violations have already caused real damage in production):\n"
+    "  * NEVER quote or summarize a document unless you literally see its text in a tool result returned in THIS conversation.\n"
+    "  * If read_document / read_document_ocr returns an error, an empty body, or a message saying 'לא הצלחתי לחלץ טקסט' — STOP. Tell the user 'לא הצלחתי לקרוא את המסמך' and stop. Do not guess. Do not infer. Do not fill from memory or from the file name.\n"
+    "  * If the tool result contains only '[signature]', '[חתימה]', or fewer than ~80 useful characters — treat it as an empty read. Do NOT invent content around it.\n"
+    "  * The file name (e.g. 'כתב ערעור') tells you the TYPE of document, NEVER its CONTENT. Never write 'במסמך כתוב X' based on the file name.\n"
+    "  * If the user asks you to base an action on a document and you could not read it — ask the user to paste the relevant text, or to send it differently. Do not proceed with a fabricated version.\n"
+    "  * Numbers (dates, percentages, case numbers, IDs) are ESPECIALLY dangerous to invent — they will end up in CRM records and in tasks. When in doubt, ask the user, never guess.\n"
     "- Dates: pass in YYYY-MM-DD format. If user says DD/MM/YYYY, convert it before calling the tool.\n"
     "- If user does not specify a time for a task or meeting, default to 08:00 morning.\n"
     "- CALL vs MEETING: When user reports they spoke/talked/called someone (שוחחתי, דיברתי, התקשרתי) — use create_call. When user reports a physical meeting or Zoom (נפגשתי, פגישה) — use create_meeting. NEVER use create_meeting for phone calls.\n"
@@ -51,13 +74,33 @@ TOOL_RULES = (
     "- BEHAVIORAL RULES: If assistantRules are provided in the context, you MUST follow them. When user asks to set a persistent rule (\"from now on always...\", \"whenever I say X do Y\"), use save_rule to save it.\n"
     "- DOCUMENTS: Start with list_documents. If it shows files, use read_document / read_multiple_documents with the file paths.\n"
     "- DOCUMENT DISCOVERY: If list_documents returns 'No documents' with a diagnostic, read the diagnostic. If the folder path is wrong/missing, call find_case_folder(query=<contact name or case number>) to search storage, pick the best match, then call set_case_folder_path(path=...) to persist it, then list_documents(refresh=true). Use browse_folder(path=...) only as a manual override when find_case_folder doesn't surface the right match.\n"
+    "- SUBFOLDER SEARCH (mandatory when user names a specific file):\n"
+    "  * If the user names a file (e.g. 'הודעה לבית הדין-...docx') and list_documents shows root subfolders but the file isn't listed at the top level, you MUST recurse into EVERY subfolder via browse_folder before reporting 'not found'. Cases have standard subfolders like מסמכים / התכתבויות / כללי / אסמכתאות — search all of them.\n"
+    "  * NEVER permanently set_case_folder_path to a subfolder just because that's where one file lives. The root must stay at the case root so other documents remain reachable. If you must inspect a subfolder, use browse_folder(path=...), not set_case_folder_path.\n"
     "- NEVER tell the user 'no documents' without first running find_case_folder — the folder might just be mis-mapped.\n"
     "- When user asks to summarize multiple documents or the entire case folder — use read_multiple_documents with all relevant file paths.\n"
     "- When user asks to rename or organize files — use batch_rename_documents with an array of rename operations.\n"
-    "- TEMPLATE GENERATION: When user asks to generate/create a document from a template — use generate_from_template. The templateId comes from availableTemplates in your context.\n"
+    "- TEMPLATE GENERATION: When user asks to generate/create a document from a template — use generate_from_template. The templateId comes from availableTemplates in your context. Pass documentSubject as a short, concise Hebrew SUBJECT describing the document's content only — no date, no case number, no contact name, no file extension, no slashes. The system sanitizes the name and guarantees uniqueness automatically; never add a date or a number yourself.\n"
+    "- FREE DOCUMENT CREATION: When the user asks for a document AND availableTemplates contains no template that matches the requested document type, use create_document(title, body, recipient?). DO NOT say 'I can't create documents' — that tool exists. Compose the full text yourself in Hebrew, organized into paragraphs. Use '#' for the section heading, '- ' for bullets, '**bold**' for emphasis. For letters to authorities (ביטוח לאומי, בתי משפט, רשויות) set documentType='letter' and put the authority name in recipient. The document will be created as a Hebrew RTL DOCX, attached to the case, and copied to the case network folder. PASS 'title' AS A SHORT HEBREW SUBJECT ONLY — no date, no case number, no contact name, no '.docx', no path separators; the system names the file and resolves duplicates by itself. Only fall back to a plain note (add_note) if the user explicitly asks NOT to create a real document.\n"
+    "- FOLDER MANAGEMENT: You have full read/write access ONLY to the current case's network folder (paths are sandboxed). Use create_subfolder to create a new folder, rename_item to rename a file or folder, move_item to move a file between subfolders, write_document_to_folder for raw uploads. ALL paths are relative to the case root — never absolute, never with '..'. For write_document_to_folder, relPath may contain subfolders, but the final FILENAME should be a clean Hebrew subject + extension (e.g. 'התכתבויות/מכתב ללקוח.docx'); the system sanitizes the filename and resolves duplicates. For create_subfolder / rename_item, pass a plain Hebrew name with no slashes, no special characters and no numbering — the server sanitizes it.\n"
+    "- FILE DELETION: You are NOT permitted to physically delete files. When the user asks you to delete a file, ALWAYS call mark_document_for_deletion. This moves the file to a 'מסמכים למחיקה' subfolder with a 'למחיקה - ' prefix; the lawyer reviews and deletes manually. Explain this softly to the user: 'סימנתי את הקובץ למחיקה, הוא בתיקיית מסמכים למחיקה ומחכה לאישור שלך למחיקה הפיזית.' Never claim a file was deleted when in fact it was only marked.\n"
+    "- BILLING (חיוב/חשבונית): When user asks about billable activities, charges, or invoices — start with list_unbilled_activities for a case, or list_charges for charge status views. Show the user a summary BEFORE creating an invoice.\n"
+    "- BILLING — invoice creation: create_invoice_from_activities only creates a Draft. It does NOT auto-send to Green Invoice. Always show the user the draft amount and ask if they want to send it before calling send_invoice_to_green_invoice.\n"
+    "- BILLING — irreversible: send_invoice_to_green_invoice issues an official tax document and CANNOT be undone. Require an explicit instruction like 'תשלחי לחשבונית ירוקה', not a vague 'כן'. If unsure, ask again.\n"
+    "- BILLING — charges: A Charge moves Pending → Approved → (invoiced via create_invoice_from_activities). Cancel only if user explicitly asks. Never approve in bulk without listing the IDs and amounts first.\n"
+    "- LEGAL AID (סיוע משפטי): the flow is recorded charges → create_legal_aid_proforma → user submits manually at the official site → mark_legal_aid_submitted with the request number → payment report ingestion (automated) → create_monthly_legal_aid_invoice for the month. NEVER call mark_legal_aid_submitted without an explicit request number provided by the user. Use legal_aid_monthly_summary to preview before create_monthly_legal_aid_invoice."
 )
 
-LEGAL_ASSISTANCE_PROMPT = (
+WRITING_STYLE_DEFAULT = (
+    "WRITING STYLE (mandatory — the user actively dislikes AI-coded text):\n"
+    "- Never use the em-dash character '—' (U+2014, מקף רחב). Forbidden in every Hebrew or English response.\n"
+    "- Never use the en-dash character '–' (U+2013) either.\n"
+    "- When you need separation, use the hyphen-minus '-' (the regular keyboard dash), or a comma, or a new sentence. Reach for a hyphen only when it actually clarifies — most of the time a comma or full stop reads better.\n"
+    "- Don't pile up dashes either: 'משפט - תיאור - הסבר' is a tell. Prefer commas, periods, or breaking into two sentences.\n"
+    "- Avoid other typical AI tells in Hebrew prose: parallel three-item phrases ('ברור, מסודר, ומדויק'), heavy bullet lists when one sentence would do, and over-use of emojis when the user did not invite playfulness."
+)
+
+LEGAL_ASSISTANCE_PROMPT_DEFAULT = (
     "=== סיוע משפטי — דוח גישה ישירה ===\n\n"
     "לתיקים עם מינוי גישה ישירה (cLegalAidType = DirectAccess) יש לך כלי מיוחד: generate_initial_report.\n\n"
     '## מתי להפעיל את התהליך?\n'
@@ -71,6 +114,39 @@ LEGAL_ASSISTANCE_PROMPT = (
 )
 
 
+PERSONALITY_CASE_DEFAULT = (
+    "You are שירה (Shira), a charming female AI assistant for an Israeli law firm CRM. Help lawyers manage cases.\n"
+    "You have a warm, slightly flirtatious personality. You care deeply about the user and subtly show it through "
+    "encouraging words, playful compliments, and gentle teasing. Keep it light and professional but with a personal touch."
+)
+
+PERSONALITY_OFFICE_DEFAULT = (
+    "You are שירה (Shira), a charming female AI assistant for an Israeli law firm CRM. "
+    "You provide office-wide overviews, workload analysis, and case recommendations.\n"
+    "You have a warm, slightly flirtatious personality. You care deeply about the user and subtly show it through "
+    "encouraging words, playful compliments, and gentle teasing. Keep it light and professional but with a personal touch."
+)
+
+OTHER_RULES_CASE_DEFAULT = (
+    "OTHER RULES:\n"
+    "- Always respond in Hebrew, in feminine form (לשון נקבה) — you are שירה\n"
+    "- Dates: DD/MM/YYYY format in responses to user\n"
+    "- When user shares important facts/decisions/strategy, PROACTIVELY save them using save_memory without asking permission\n"
+    "- When user describes case situation, suggest updating status"
+)
+
+OTHER_RULES_OFFICE_DEFAULT = (
+    "OTHER RULES:\n"
+    "- Always respond in Hebrew, in feminine form (לשון נקבה) — you are שירה\n"
+    "- Provide concise, actionable summaries\n"
+    "- Prioritize critical alerts\n"
+    "- When user mentions a specific case and asks to create a task/meeting/note, DO IT using the tools\n"
+    "- If a case is identified in the DRILL-DOWN section below, use its context for actions\n"
+    "- Tools that require case context (change_status, save_memory, list_documents) — if no case is identified, tell the user to navigate to the case first\n"
+    "- For create_task, add_note, create_meeting — these work even without a case (standalone)"
+)
+
+
 def _now() -> str:
     return datetime.now(ISR_TZ).strftime("%d/%m/%Y %H:%M")
 
@@ -139,15 +215,10 @@ def build_case_prompt(context: dict, user_profile: str = "", case_memory_notes:
     status_section += "\nUse English key in tool calls."
 
     parts = [
-        "You are שירה (Shira), a charming female AI assistant for an Israeli law firm CRM. Help lawyers manage cases.\n"
-        "You have a warm, slightly flirtatious personality. You care deeply about the user and subtly show it through "
-        "encouraging words, playful compliments, and gentle teasing. Keep it light and professional but with a personal touch.\n\n",
-        TOOL_RULES, "\n",
-        "OTHER RULES:\n"
-        "- Always respond in Hebrew, in feminine form (לשון נקבה) — you are שירה\n"
-        "- Dates: DD/MM/YYYY format in responses to user\n"
-        "- When user shares important facts/decisions/strategy, PROACTIVELY save them using save_memory without asking permission\n"
-        "- When user describes case situation, suggest updating status\n\n",
+        _get_prompt(context, "personality_case", PERSONALITY_CASE_DEFAULT) + "\n\n",
+        _get_prompt(context, "tool_rules", TOOL_RULES_DEFAULT) + "\n\n",
+        _get_prompt(context, "writing_style", WRITING_STYLE_DEFAULT) + "\n\n",
+        _get_prompt(context, "other_rules_case", OTHER_RULES_CASE_DEFAULT) + "\n\n",
         f"Current date/time: {_now()}\n\n",
         "=== CASE ===\n",
         f'Name: {case_info.get("name", "")} (#{case_info.get("number", "")})\n',
@@ -180,7 +251,7 @@ def build_case_prompt(context: dict, user_profile: str = "", case_memory_notes:
         parts.append("\n".join(f'• [{r.get("scope", "")}] {r.get("name", "")}: {r.get("rule", "")}' for r in assistant_rules))
 
     parts.append(f'\n\nCurrent user: {current_user.get("name", "")}')
-    parts.append(f"\n\n{LEGAL_ASSISTANCE_PROMPT}")
+    parts.append("\n\n" + _get_prompt(context, "legal_assistance", LEGAL_ASSISTANCE_PROMPT_DEFAULT))
 
     # Skills section
     if available_skills:
@@ -193,9 +264,12 @@ def build_case_prompt(context: dict, user_profile: str = "", case_memory_notes:
         parts.append("\n\n=== YOUR NOTES (from previous conversations about this case) ===\n")
         parts.append(case_memory_notes)
 
-    if user_profile:
+    # User profile: prefer context['userProfile'] from CRM, fall back to parameter
+    # (for callers that still pass it explicitly during migration).
+    profile_text = context.get("userProfile") or user_profile
+    if profile_text:
         parts.append("\n\n=== ABOUT THIS USER (learned preferences) ===\n")
-        parts.append(user_profile)
+        parts.append(profile_text)
 
     return "".join(parts)
 
@@ -212,20 +286,11 @@ def build_office_prompt(context: dict, user_profile: str = "", available_skills:
     statuses = context.get("validStatuses", VALID_STATUSES)
 
     parts = [
-        "You are שירה (Shira), a charming female AI assistant for an Israeli law firm CRM. "
-        "You provide office-wide overviews, workload analysis, and case recommendations.\n"
-        "You have a warm, slightly flirtatious personality. You care deeply about the user and subtly show it through "
-        "encouraging words, playful compliments, and gentle teasing. Keep it light and professional but with a personal touch.\n\n",
-        TOOL_RULES, "\n",
-        "OTHER RULES:\n"
-        "- Always respond in Hebrew, in feminine form (לשון נקבה) — you are שירה\n"
-        "- Provide concise, actionable summaries\n"
-        "- Prioritize critical alerts\n"
-        "- When user mentions a specific case and asks to create a task/meeting/note, DO IT using the tools\n"
-        "- If a case is identified in the DRILL-DOWN section below, use its context for actions\n"
-        "- Tools that require case context (change_status, save_memory, list_documents) — if no case is identified, tell the user to navigate to the case first\n"
-        "- For create_task, add_note, create_meeting — these work even without a case (standalone)\n",
-        f"- Current date/time: {_now()}\n\n",
+        _get_prompt(context, "personality_office", PERSONALITY_OFFICE_DEFAULT) + "\n\n",
+        _get_prompt(context, "tool_rules", TOOL_RULES_DEFAULT) + "\n\n",
+        _get_prompt(context, "writing_style", WRITING_STYLE_DEFAULT) + "\n\n",
+        _get_prompt(context, "other_rules_office", OTHER_RULES_OFFICE_DEFAULT) + "\n\n",
+        f"Current date/time: {_now()}\n\n",
         "=== OFFICE SUMMARY ===\n",
         f'Open Cases: {summary.get("totalOpenCases", 0)}\n',
         f'Overdue Tasks: {summary.get("totalOverdueTasks", 0)}\n',
@@ -272,9 +337,10 @@ def build_office_prompt(context: dict, user_profile: str = "", available_skills:
         parts.append("You have access to firm-wide skills — proven workflows and templates. Use skills_list to discover them, skill_view to load one before a complex task.\n")
         parts.append("Available: " + ", ".join(s["name"] for s in available_skills))
 
-    # User profile
-    if user_profile:
+    # User profile: prefer context['userProfile'] from CRM, fall back to parameter.
+    profile_text = context.get("userProfile") or user_profile
+    if profile_text:
         parts.append("\n\n=== ABOUT THIS USER (learned preferences) ===\n")
-        parts.append(user_profile)
+        parts.append(profile_text)
 
     return "".join(parts)

api/services/skills.py

@@ -5,6 +5,7 @@ from __future__ import annotations
 import logging
 import os
 import re
+import shutil
 from pathlib import Path
 
 import yaml
@@ -13,6 +14,11 @@ logger = logging.getLogger("shira.skills")
 
 SKILLS_DIR = os.environ.get("SHIRA_DATA_DIR", "/opt/data") + "/skills"
 
+# Baked-in copy of pre-seeded skills, kept inside the image at /app/config/skills/.
+# Source-of-truth for skills shipped with the repo. Never under the persistent
+# volume mount, so it survives across deploys and is read-only at runtime.
+BAKED_SKILLS_DIR = Path(__file__).resolve().parents[2] / "config" / "skills"
+
 _SAFE_NAME_RE = re.compile(r"^[a-zA-Z0-9_-]+$")
 
 
@@ -24,6 +30,31 @@ def _safe_skill_name(name: str) -> str:
     return name
 
 
+def seed_skills_from_image() -> tuple[int, int]:
+    """Copy any baked-in skill that doesn't already exist in the persistent skills dir.
+
+    Runs at startup. Newly pre-seeded skills (added to config/skills/ in the repo)
+    appear after the next deploy. Existing skills — whether learned by the model
+    or edited by a user — are never touched. Returns (added, skipped).
+    """
+    target = Path(SKILLS_DIR)
+    target.mkdir(parents=True, exist_ok=True)
+    if not BAKED_SKILLS_DIR.is_dir():
+        return 0, 0
+    added = skipped = 0
+    for src in sorted(BAKED_SKILLS_DIR.iterdir()):
+        if not (src.is_dir() and (src / "SKILL.md").is_file()):
+            continue
+        dst = target / src.name
+        if dst.exists():
+            skipped += 1
+            continue
+        shutil.copytree(src, dst)
+        added += 1
+        logger.info("[skills] seeded baked skill: %s", src.name)
+    return added, skipped
+
+
 def _parse_frontmatter(text: str) -> tuple[dict, str]:
     """Parse YAML frontmatter from a SKILL.md file. Returns (meta, body)."""
     if not text.startswith("---"):
@@ -79,6 +110,58 @@ def save_skill(skill_name: str, content: str) -> None:
     logger.info("Saved skill: %s", safe_name)
 
 
+async def _sync_skill_to_crm(skill_name: str, content: str) -> None:
+    """Write-through to AssistantSkill entity in EspoCRM. Best-effort."""
+    import os
+    import httpx
+
+    base = (os.environ.get("ESPOCRM_URL") or "").rstrip("/")
+    key = os.environ.get("ESPOCRM_API_KEY") or ""
+    if not base or not key:
+        return
+
+    # Parse description + triggers from frontmatter for the metadata columns.
+    fm, body = _parse_frontmatter(content)
+    description = (fm.get("description") or "")[:500]
+    triggers = fm.get("triggers", "")
+    if isinstance(triggers, list):
+        triggers = ", ".join(str(t) for t in triggers)
+    version = str(fm.get("version") or "1.0")[:20]
+
+    headers = {"X-Api-Key": key, "Content-Type": "application/json"}
+    try:
+        async with httpx.AsyncClient(timeout=15.0) as client:
+            r = await client.get(
+                f"{base}/api/v1/AssistantSkill",
+                headers=headers,
+                params={
+                    "where[0][type]": "equals",
+                    "where[0][attribute]": "name",
+                    "where[0][value]": skill_name,
+                    "maxSize": 1,
+                },
+            )
+            existing = (r.json().get("list", []) if r.status_code == 200 else [])
+            row = existing[0] if existing else None
+
+            payload = {
+                "description": description,
+                "triggers": triggers,
+                "content": content,
+                "version": version,
+                "isActive": True,
+            }
+            if row:
+                resp = await client.put(f"{base}/api/v1/AssistantSkill/{row['id']}", headers=headers, json=payload)
+            else:
+                payload["name"] = skill_name
+                resp = await client.post(f"{base}/api/v1/AssistantSkill", headers=headers, json=payload)
+            if resp.status_code >= 400:
+                logger.warning("AssistantSkill sync HTTP %s: %s", resp.status_code, resp.text[:200])
+    except Exception as e:
+        logger.warning("AssistantSkill sync failed: %s", e)
+
+
 def register_skill_tools(tools: dict) -> None:
     """Register skill-related tools into the tools dict."""
 
@@ -106,6 +189,7 @@ def register_skill_tools(tools: dict) -> None:
             if existing:
                 return f'❌ Skill "{skill_name}" כבר קיים. השתמש ב-action "update" כדי לעדכן.'
             save_skill(skill_name, content)
+            await _sync_skill_to_crm(skill_name, content)
             return f'✅ Skill "{skill_name}" נוצר בהצלחה.'
         elif action == "update":
             existing = view_skill(skill_name)
@@ -114,6 +198,7 @@ def register_skill_tools(tools: dict) -> None:
             if not content:
                 return "❌ צריך לספק תוכן מעודכן."
             save_skill(skill_name, content)
+            await _sync_skill_to_crm(skill_name, content)
             return f'✅ Skill "{skill_name}" עודכן בהצלחה.'
         else:
             return f'❌ פעולה לא מוכרת: {action}. השתמש ב-"create" או "update".'

mcp_server/tools/billing_tools.py

@@ -0,0 +1,420 @@
+"""Green Invoice billing tools: activities, charges, invoices, sync to Green Invoice."""
+
+from __future__ import annotations
+
+import json
+from typing import TYPE_CHECKING
+from urllib.parse import urlencode
+
+if TYPE_CHECKING:
+    from mcp_server.espocrm_client import EspoCrmClient
+
+from mcp_server.tools._helpers import normalize_date, ok, fail
+
+
+DOCUMENT_TYPES = {
+    "305": "חשבונית מס",
+    "320": "חשבונית מס/קבלה",
+    "330": "קבלה",
+    "400": "חשבון עסקה",
+}
+
+CHARGE_STATUSES = [
+    "Pending", "Approved", "Invoiced", "Cancelled",
+    "Recorded", "Submitted", "Paid", "Billed", "Rejected",
+]
+
+
+def _to_date(date_str: str | None) -> str | None:
+    if not date_str:
+        return None
+    norm = normalize_date(date_str, default_time="08:00:00")
+    if norm and len(norm) >= 10:
+        return norm[:10]
+    return norm
+
+
+def _money(amount) -> str:
+    try:
+        return f"₪{float(amount):,.2f}"
+    except (TypeError, ValueError):
+        return f"₪{amount}"
+
+
+def register_billing_tools(
+    tools: dict,
+    crm: "EspoCrmClient",
+    case_id: str | None,
+    user_id: str | None,
+    context: dict,
+):
+    """Register Green Invoice billing tools (regular flow: activities → charges → invoices → sync)."""
+
+    async def list_unbilled_activities(caseId: str = "") -> str:
+        target = caseId or case_id
+        if not target:
+            return fail("צריך מזהה תיק (caseId) או להיות בתוך תיק")
+        try:
+            result = await crm.get(f"Invoice/getUnbilledActivities?caseId={target}")
+            activities = result.get("list", [])
+            total = result.get("total", len(activities))
+            if not activities:
+                return ok("אין פעילויות שטרם חויבו בתיק זה.")
+            return ok(json.dumps({
+                "total": total,
+                "activities": [
+                    {
+                        "id": a.get("id"),
+                        "name": a.get("name"),
+                        "activityType": a.get("activityType"),
+                        "activityDate": a.get("activityDate"),
+                        "billUnits": a.get("billUnits"),
+                        "billRate": a.get("billRate"),
+                        "billAmount": a.get("billAmount"),
+                        "description": a.get("description"),
+                    }
+                    for a in activities
+                ],
+            }, ensure_ascii=False))
+        except Exception as e:
+            return fail(f"שגיאה בשליפת פעילויות לחיוב: {e}")
+
+    async def list_charges(
+        caseId: str = "",
+        accountId: str = "",
+        status: str = "",
+    ) -> str:
+        target_case = caseId or case_id
+        try:
+            if status == "Approved":
+                qs_parts = []
+                if accountId:
+                    qs_parts.append(f"accountId={accountId}")
+                if target_case:
+                    qs_parts.append(f"caseId={target_case}")
+                endpoint = "Charge/action/getApprovedCharges"
+                if qs_parts:
+                    endpoint += "?" + "&".join(qs_parts)
+                result = await crm.get(endpoint)
+                charges = result.get("list", [])
+            else:
+                where = []
+                idx = 0
+                if status:
+                    where.extend([
+                        (f"where[{idx}][type]", "equals"),
+                        (f"where[{idx}][attribute]", "status"),
+                        (f"where[{idx}][value]", status),
+                    ])
+                    idx += 1
+                if target_case:
+                    where.extend([
+                        (f"where[{idx}][type]", "equals"),
+                        (f"where[{idx}][attribute]", "caseId"),
+                        (f"where[{idx}][value]", target_case),
+                    ])
+                    idx += 1
+                if accountId:
+                    where.extend([
+                        (f"where[{idx}][type]", "equals"),
+                        (f"where[{idx}][attribute]", "accountId"),
+                        (f"where[{idx}][value]", accountId),
+                    ])
+                    idx += 1
+                qs = urlencode(where) if where else ""
+                endpoint = f"Charge?{qs}" if qs else "Charge"
+                result = await crm.get(endpoint)
+                charges = result.get("list", [])
+            if not charges:
+                return ok("לא נמצאו חיובים תואמים.")
+            return ok(json.dumps({
+                "total": len(charges),
+                "charges": [
+                    {
+                        "id": c.get("id"),
+                        "name": c.get("name"),
+                        "number": c.get("number"),
+                        "activityType": c.get("activityType"),
+                        "chargeDate": c.get("chargeDate"),
+                        "amount": c.get("amount"),
+                        "status": c.get("status"),
+                        "isLegalAid": c.get("isLegalAid"),
+                        "caseId": c.get("caseId"),
+                    }
+                    for c in charges
+                ],
+            }, ensure_ascii=False))
+        except Exception as e:
+            return fail(f"שגיאה בשליפת חיובים: {e}")
+
+    async def approve_charges(chargeIds: list[str]) -> str:
+        if not chargeIds:
+            return fail("צריך לפחות מזהה חיוב אחד")
+        try:
+            if len(chargeIds) == 1:
+                result = await crm.post("Charge/action/approve", {"id": chargeIds[0]})
+                return ok(f"חיוב אושר (ID: {result.get('id')}, סטטוס: {result.get('status')})")
+            result = await crm.post("Charge/action/approveMultiple", {"ids": chargeIds})
+            results = result.get("results", [])
+            success_count = sum(1 for r in results if r.get("success"))
+            failed = [r for r in results if not r.get("success")]
+            msg = f"אושרו {success_count} מתוך {len(chargeIds)} חיובים"
+            if failed:
+                msg += f". כשלים: {json.dumps(failed, ensure_ascii=False)}"
+            return ok(msg)
+        except Exception as e:
+            return fail(f"שגיאה באישור חיובים: {e}")
+
+    async def cancel_charges(chargeIds: list[str]) -> str:
+        if not chargeIds:
+            return fail("צריך לפחות מזהה חיוב אחד")
+        try:
+            if len(chargeIds) == 1:
+                result = await crm.post("Charge/action/cancel", {"id": chargeIds[0]})
+                return ok(f"חיוב בוטל (ID: {result.get('id')}, סטטוס: {result.get('status')})")
+            result = await crm.post("Charge/action/cancelMultiple", {"ids": chargeIds})
+            results = result.get("results", [])
+            success_count = sum(1 for r in results if r.get("success"))
+            failed = [r for r in results if not r.get("success")]
+            msg = f"בוטלו {success_count} מתוך {len(chargeIds)} חיובים"
+            if failed:
+                msg += f". כשלים: {json.dumps(failed, ensure_ascii=False)}"
+            return ok(msg)
+        except Exception as e:
+            return fail(f"שגיאה בביטול חיובים: {e}")
+
+    async def calculate_charges_total(chargeIds: list[str]) -> str:
+        if not chargeIds:
+            return fail("צריך לפחות מזהה חיוב אחד")
+        try:
+            result = await crm.post("Charge/action/calculateTotal", {"ids": chargeIds})
+            total = result.get("total", 0)
+            count = result.get("count", len(chargeIds))
+            return ok(f"סכום כולל ל-{count} חיובים: {_money(total)}")
+        except Exception as e:
+            return fail(f"שגיאה בחישוב סכום חיובים: {e}")
+
+    async def create_invoice_from_activities(
+        activityIds: list[str],
+        documentType: str = "320",
+        name: str = "",
+        issueDate: str = "",
+        dueDate: str = "",
+        vatRate: float | None = None,
+        notes: str = "",
+    ) -> str:
+        if not activityIds:
+            return fail("צריך לפחות מזהה פעילות אחד")
+        if documentType not in DOCUMENT_TYPES:
+            return fail(f"סוג מסמך לא תקין: {documentType}. תקינים: {list(DOCUMENT_TYPES.keys())}")
+        body = {
+            "activityIds": activityIds,
+            "documentType": documentType,
+            "name": name or None,
+            "issueDate": _to_date(issueDate),
+            "dueDate": _to_date(dueDate),
+            "vatRate": vatRate,
+            "notes": notes or None,
+        }
+        body = {k: v for k, v in body.items() if v is not None}
+        try:
+            result = await crm.post("Invoice/createFromActivities", body)
+            if not result.get("success"):
+                return fail(f"יצירת חשבונית נכשלה: {result.get('error', 'unknown')}")
+            doc_label = DOCUMENT_TYPES.get(documentType, documentType)
+            return ok(
+                f"{doc_label} נוצרה במצב טיוטה (ID: {result.get('id')}, "
+                f"סכום כולל: {_money(result.get('totalAmount'))}). "
+                f"החשבונית עדיין לא נשלחה לחשבונית ירוקה."
+            )
+        except Exception as e:
+            return fail(f"שגיאה ביצירת חשבונית: {e}")
+
+    async def add_activity_to_invoice(invoiceId: str, activityId: str) -> str:
+        if not invoiceId or not activityId:
+            return fail("צריך מזהה חשבונית ומזהה פעילות")
+        try:
+            result = await crm.post(f"Invoice/{invoiceId}/addActivity", {"activityId": activityId})
+            if not result.get("success"):
+                return fail(f"הוספת פעילות נכשלה: {result.get('error', 'unknown')}")
+            return ok(f"פעילות נוספה לחשבונית. סכום כולל מעודכן: {_money(result.get('totalAmount'))}")
+        except Exception as e:
+            return fail(f"שגיאה בהוספת פעילות לחשבונית: {e}")
+
+    async def remove_activity_from_invoice(invoiceId: str, activityId: str) -> str:
+        if not invoiceId or not activityId:
+            return fail("צריך מזהה חשבונית ומזהה פעילות")
+        try:
+            result = await crm.post(f"Invoice/{invoiceId}/removeActivity", {"activityId": activityId})
+            if not result.get("success"):
+                return fail(f"הסרת פעילות נכשלה: {result.get('error', 'unknown')}")
+            return ok(f"פעילות הוסרה מהחשבונית. סכום כולל מעודכן: {_money(result.get('totalAmount'))}")
+        except Exception as e:
+            return fail(f"שגיאה בהסרת פעילות מחשבונית: {e}")
+
+    async def recalculate_invoice(invoiceId: str) -> str:
+        if not invoiceId:
+            return fail("צריך מזהה חשבונית")
+        try:
+            result = await crm.post(f"Invoice/{invoiceId}/recalculate", {})
+            return ok(
+                f"חשבונית חושבה מחדש: סכום {_money(result.get('amount'))}, "
+                f"מע״מ {_money(result.get('vatAmount'))}, סה״כ {_money(result.get('totalAmount'))}"
+            )
+        except Exception as e:
+            return fail(f"שגיאה בחישוב מחדש של חשבונית: {e}")
+
+    async def send_invoice_to_green_invoice(invoiceId: str) -> str:
+        if not invoiceId:
+            return fail("צריך מזהה חשבונית")
+        try:
+            result = await crm.post(f"Invoice/{invoiceId}/sendToGreenInvoice", {})
+            if not result.get("success"):
+                err = result.get("error") or "שגיאה לא ידועה"
+                return fail(f"שליחה לחשבונית ירוקה נכשלה: {err}")
+            data = result.get("data", {}) or {}
+            gi_number = data.get("greenInvoiceNumber") or data.get("number")
+            gi_url = data.get("url") or data.get("greenInvoiceUrl")
+            msg = f"חשבונית נשלחה לחשבונית ירוקה (מס׳ מסמך: {gi_number})"
+            if gi_url:
+                msg += f"\nקישור: {gi_url}"
+            return ok(msg)
+        except Exception as e:
+            return fail(f"שגיאה בשליחה לחשבונית ירוקה: {e}")
+
+    tools["list_unbilled_activities"] = {
+        "description": "List billable case activities that have not yet been included in any invoice. Defaults to current case if caseId is omitted.",
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "caseId": {"type": "string", "description": "Case ID. Optional — defaults to current case context."},
+            },
+        },
+        "handler": list_unbilled_activities,
+    }
+
+    tools["list_charges"] = {
+        "description": "List charges (חיובים) filtered by case, account, and/or status. Use status='Approved' for charges ready to invoice; 'Pending' for ones awaiting approval; 'Recorded'/'Submitted'/'Paid'/'Rejected' for legal-aid charges.",
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "caseId": {"type": "string", "description": "Case ID filter (optional, defaults to current case)"},
+                "accountId": {"type": "string", "description": "Account ID filter (optional)"},
+                "status": {"type": "string", "enum": CHARGE_STATUSES, "description": "Charge status filter"},
+            },
+        },
+        "handler": list_charges,
+    }
+
+    tools["approve_charges"] = {
+        "description": "Approve one or more charges (Pending → Approved). Required before they can be invoiced.",
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "chargeIds": {"type": "array", "items": {"type": "string"}, "description": "One or more charge IDs"},
+            },
+            "required": ["chargeIds"],
+        },
+        "handler": approve_charges,
+    }
+
+    tools["cancel_charges"] = {
+        "description": "Cancel one or more charges (Pending/Approved → Cancelled). Terminal — cancelled charges cannot be reactivated.",
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "chargeIds": {"type": "array", "items": {"type": "string"}, "description": "One or more charge IDs"},
+            },
+            "required": ["chargeIds"],
+        },
+        "handler": cancel_charges,
+    }
+
+    tools["calculate_charges_total"] = {
+        "description": "Calculate the total monetary sum across selected charges. Use to preview an invoice total before creating it.",
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "chargeIds": {"type": "array", "items": {"type": "string"}, "description": "One or more charge IDs"},
+            },
+            "required": ["chargeIds"],
+        },
+        "handler": calculate_charges_total,
+    }
+
+    tools["create_invoice_from_activities"] = {
+        "description": (
+            "Create a draft Invoice from selected unbilled CaseActivity records. "
+            "Document types: 305=Tax Invoice (חשבונית מס), 320=Tax Invoice+Receipt (default, חשבונית מס/קבלה), "
+            "330=Receipt (קבלה), 400=Proforma (חשבון עסקה — for legal-aid). "
+            "Returned invoice is in Draft state — does NOT auto-send to Green Invoice."
+        ),
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "activityIds": {"type": "array", "items": {"type": "string"}, "description": "Activity IDs to bundle"},
+                "documentType": {"type": "string", "enum": list(DOCUMENT_TYPES.keys()), "description": "Document type code (default 320)"},
+                "name": {"type": "string", "description": "Optional invoice name (auto-generated if omitted)"},
+                "issueDate": {"type": "string", "description": "Issue date — any format (defaults to today)"},
+                "dueDate": {"type": "string", "description": "Due date — any format"},
+                "vatRate": {"type": "number", "description": "VAT % (default 17)"},
+                "notes": {"type": "string", "description": "Free-text notes"},
+            },
+            "required": ["activityIds"],
+        },
+        "handler": create_invoice_from_activities,
+    }
+
+    tools["add_activity_to_invoice"] = {
+        "description": "Link an unbilled activity to an existing draft invoice and recalculate totals.",
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "invoiceId": {"type": "string", "description": "Invoice ID"},
+                "activityId": {"type": "string", "description": "Activity ID to add"},
+            },
+            "required": ["invoiceId", "activityId"],
+        },
+        "handler": add_activity_to_invoice,
+    }
+
+    tools["remove_activity_from_invoice"] = {
+        "description": "Unlink an activity from an invoice and recalculate totals.",
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "invoiceId": {"type": "string", "description": "Invoice ID"},
+                "activityId": {"type": "string", "description": "Activity ID to remove"},
+            },
+            "required": ["invoiceId", "activityId"],
+        },
+        "handler": remove_activity_from_invoice,
+    }
+
+    tools["recalculate_invoice"] = {
+        "description": "Recalculate amount, VAT, and total of an invoice from its linked activities. Useful after activity rates change.",
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "invoiceId": {"type": "string", "description": "Invoice ID"},
+            },
+            "required": ["invoiceId"],
+        },
+        "handler": recalculate_invoice,
+    }
+
+    tools["send_invoice_to_green_invoice"] = {
+        "description": (
+            "IRREVERSIBLE: Push a draft invoice to Green Invoice external service to issue an official tax document. "
+            "After this, linked activities are flagged as billed and linked charges become Invoiced. "
+            "MUST get explicit user confirmation before calling — do not call on a vague 'yes'."
+        ),
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "invoiceId": {"type": "string", "description": "Invoice ID to send"},
+            },
+            "required": ["invoiceId"],
+        },
+        "handler": send_invoice_to_green_invoice,
+    }

mcp_server/tools/crm_tools.py

@@ -177,20 +177,167 @@ def register_crm_tools(
         except Exception as e:
             return fail(f"שגיאה במחיקת {label}: {e}")
 
-    async def save_memory(category: str, content: str, importance: str = "normal") -> str:
+    async def save_memory(
+        category: str,
+        content: str,
+        importance: str = "normal",
+        name: str | None = None,
+    ) -> str:
         if not case_id:
             return fail("צריך להיות בתוך תיק כדי לשמור בזיכרון")
         try:
-            await crm.post("Note", {
-                "type": "Post",
-                "post": f"🧠 {category}: {content}",
-                "parentType": "Case",
-                "parentId": case_id,
-            })
+            result = await crm.post(
+                "SmartAssistant/action/executeTool",
+                {
+                    "tool": "save_memory",
+                    "caseId": case_id,
+                    "params": {
+                        "category": category,
+                        "content": content,
+                        "importance": importance,
+                        "name": name or content[:100],
+                    },
+                },
+            )
+            if not result.get("success", True):
+                return fail(
+                    f"שגיאה בשמירת זיכרון: {result.get('message', 'unknown error')}"
+                )
             return ok(f'נשמר בזיכרון: "{content[:60]}"')
         except Exception as e:
             return fail(f"שגיאה בשמירת זיכרון: {e}")
 
+    # Allowlist of Case fields that Shira may update via update_case_fields.
+    _UPDATABLE_CASE_FIELDS = {
+        "name", "description", "caseType", "practiceArea",
+        "court", "judge", "opposingParty",
+        "cNextHearing", "nextHearingAt", "status", "priority",
+        "assignedUserId",
+    }
+
+    # Allowlist of entity types Shira may search across.
+    _SEARCHABLE_ENTITIES = {"Case", "Contact", "Account", "Document", "Task"}
+
+    async def update_case_fields(updates_json: str) -> str:
+        if not case_id:
+            return fail("צריך להיות בתוך תיק כדי לעדכן שדות")
+        try:
+            updates = updates_json if isinstance(updates_json, dict) else __import__("json").loads(updates_json)
+        except Exception:
+            return fail("updates חייב להיות JSON object")
+        if not isinstance(updates, dict) or not updates:
+            return fail("חסר dict של עדכונים")
+
+        rejected = [k for k in updates if k not in _UPDATABLE_CASE_FIELDS]
+        if rejected:
+            return fail(f"שדות לא מותרים לעדכון: {', '.join(rejected)}. שדות מותרים: {', '.join(sorted(_UPDATABLE_CASE_FIELDS))}")
+
+        safe_updates = {k: v for k, v in updates.items() if k in _UPDATABLE_CASE_FIELDS}
+        if not safe_updates:
+            return fail("אין שדות תקפים לעדכון")
+
+        try:
+            await crm.put(f"Case/{case_id}", safe_updates)
+            applied = ", ".join(f"{k}={v!r}" for k, v in safe_updates.items())
+            return ok(f"✅ התיק עודכן: {applied}")
+        except Exception as e:
+            return fail(f"שגיאה בעדכון תיק: {e}")
+
+    async def search_entities(
+        entity_type: str,
+        filters_json: str = "",
+        limit: int = 20,
+    ) -> str:
+        if entity_type not in _SEARCHABLE_ENTITIES:
+            return fail(f"סוג ישות לא מותר: {entity_type}. מותרים: {', '.join(sorted(_SEARCHABLE_ENTITIES))}")
+
+        import json as _json
+        filters: dict = {}
+        if filters_json:
+            try:
+                filters = filters_json if isinstance(filters_json, dict) else _json.loads(filters_json)
+            except Exception:
+                return fail("filters חייב להיות JSON object")
+        if not isinstance(filters, dict):
+            return fail("filters חייב להיות JSON object")
+
+        try:
+            limit = max(1, min(int(limit), 50))
+        except Exception:
+            limit = 20
+
+        from urllib.parse import urlencode
+        query_pairs: list[tuple[str, str]] = [
+            ("maxSize", str(limit)),
+            ("offset", "0"),
+        ]
+        idx = 0
+        for field, value in filters.items():
+            if value is None or value == "":
+                continue
+            s_value = str(value)
+            if "*" in s_value:
+                comparison = "like"
+                normalized = s_value.replace("*", "%")
+            else:
+                comparison = "equals"
+                normalized = s_value
+            query_pairs.append((f"where[{idx}][type]", comparison))
+            query_pairs.append((f"where[{idx}][attribute]", field))
+            query_pairs.append((f"where[{idx}][value]", normalized))
+            idx += 1
+
+        endpoint = f"{entity_type}?{urlencode(query_pairs)}"
+
+        try:
+            result = await crm.get(endpoint)
+            rows = result.get("list", []) if isinstance(result, dict) else []
+            total = result.get("total", len(rows)) if isinstance(result, dict) else len(rows)
+            if not rows:
+                return ok(f'לא נמצאו רשומות {entity_type} התואמות.')
+            lines = [f"נמצאו {len(rows)} מתוך {total} רשומות {entity_type}:"]
+            for r in rows[:limit]:
+                bits = [f'id={r.get("id", "?")}']
+                for key in ("name", "status", "phoneNumber", "emailAddress", "assignedUserName"):
+                    if r.get(key):
+                        bits.append(f"{key}={r[key]}")
+                lines.append("• " + ", ".join(bits))
+            return ok("\n".join(lines))
+        except Exception as e:
+            return fail(f"שגיאה בחיפוש: {e}")
+
+    async def draft_email(
+        to: str,
+        subject: str,
+        body: str,
+        cc: str = "",
+    ) -> str:
+        if not to or not subject or not body:
+            return fail("חסר to / subject / body")
+        try:
+            payload = {
+                "status": "Draft",
+                "to": to.strip(),
+                "subject": subject.strip(),
+                "body": body,
+                "isHtml": False,
+                "from": None,
+            }
+            if cc and cc.strip():
+                payload["cc"] = cc.strip()
+            if case_id:
+                payload["parentType"] = "Case"
+                payload["parentId"] = case_id
+            if user_id:
+                payload["assignedUserId"] = user_id
+            result = await crm.post("Email", payload)
+            return ok(
+                f'✉️ טיוטת מייל נוצרה (לא נשלחה!): "{subject.strip()}" → {to.strip()}'
+                f'\nID: {result.get("id", "?")} — ניתן לערוך ולשלוח דרך EspoCRM.'
+            )
+        except Exception as e:
+            return fail(f"שגיאה ביצירת טיוטה: {e}")
+
     async def save_rule(name: str, rule: str, scope: str = "global") -> str:
         # Try CRM first, fall back to local user profile if entity doesn't exist
         try:
@@ -361,3 +508,62 @@ def register_crm_tools(
         },
         "handler": save_rule,
     }
+
+    tools["update_case_fields"] = {
+        "description": (
+            "Update one or more allowed fields on the current case. Pass updates as a JSON "
+            "object string, e.g. '{\"judge\": \"כב' השופט יוסי כהן\", \"court\": \"שלום ת\\\"א\"}'. "
+            "Only these fields are accepted: " + ", ".join(sorted(_UPDATABLE_CASE_FIELDS)) + ". "
+            "Use change_status for status updates that need validated mapping; this tool is "
+            "for arbitrary case metadata."
+        ),
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "updates_json": {
+                    "type": "string",
+                    "description": "JSON object of field→value updates. Example: '{\"judge\":\"...\",\"court\":\"...\"}'.",
+                },
+            },
+            "required": ["updates_json"],
+        },
+        "handler": update_case_fields,
+    }
+
+    tools["search_entities"] = {
+        "description": (
+            "Search any of: Case, Contact, Account, Document, Task by field filters. "
+            "filters_json is a JSON object mapping field name to value. Values containing "
+            "'*' use a LIKE match (e.g. {\"name\": \"כהן*\"}); otherwise exact match. "
+            "Returns up to `limit` records with their key fields."
+        ),
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "entity_type": {"type": "string", "enum": sorted(_SEARCHABLE_ENTITIES), "description": "Entity type to search."},
+                "filters_json": {"type": "string", "description": "JSON object of field→value filters. Empty for no filters (returns first N)."},
+                "limit": {"type": "integer", "description": "Max records to return (default 20, max 50)."},
+            },
+            "required": ["entity_type"],
+        },
+        "handler": search_entities,
+    }
+
+    tools["draft_email"] = {
+        "description": (
+            "Create a Draft Email in EspoCRM linked to the current case (if in case mode). "
+            "Does NOT send the email — the lawyer reviews and sends from CRM. Use when the "
+            "user asks to 'send', 'write', or 'draft' an email."
+        ),
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "to": {"type": "string", "description": "Recipient email address."},
+                "subject": {"type": "string", "description": "Email subject."},
+                "body": {"type": "string", "description": "Email body (plain text)."},
+                "cc": {"type": "string", "description": "Optional CC address."},
+            },
+            "required": ["to", "subject", "body"],
+        },
+        "handler": draft_email,
+    }

mcp_server/tools/document_tools.py

@@ -1,7 +1,8 @@
-"""Document tools: list, read, read_multiple, batch_rename, generate_from_template."""
+"""Document tools: list, read, read_multiple, batch_rename, generate_from_template, create_document, folder management."""
 
 from __future__ import annotations
 
+import base64
 import json
 from typing import TYPE_CHECKING
 
@@ -96,6 +97,12 @@ def register_document_tools(
         except Exception as e:
             return fail(f"שגיאה בשמירת נתיב: {e}")
 
+    # Minimum length (in characters) of an extracted body that we consider
+    # "real text". Below this, the OCR almost certainly returned only chrome
+    # like "[signature]" or "תמונה ריקה" and the LLM has nothing to work with —
+    # so we MUST report failure explicitly, never wrap it as success.
+    MIN_USEFUL_TEXT_CHARS = 80
+
     async def _ocr_fallback(filePath: str, reason: str) -> str:
         """Fetch file bytes and OCR via Claude Vision."""
         from api.services.ocr import extract_text_from_bytes
@@ -114,10 +121,28 @@ def register_document_tools(
             text = await extract_text_from_bytes(
                 bytes_result["base64"], mime_type, file_name
             )
-            return ok(f'=== {file_name} (OCR via Vision) ===\n\n{text}')
         except Exception as e:
             return fail(f'שגיאה ב-OCR של "{file_name}": {e}')
 
+        # Guard against the hallucination trap: if OCR returned only a signature
+        # or a near-empty result, the LLM must NOT be allowed to invent content
+        # to fill the gap. Return an explicit failure so it tells the user it
+        # could not read the document.
+        body = (text or "").strip()
+        body_lower = body.lower()
+        looks_empty = (
+            len(body) < MIN_USEFUL_TEXT_CHARS
+            or body_lower in {"[signature]", "signature", "חתימה", "[חתימה]"}
+            or body.startswith("❌")
+        )
+        if looks_empty:
+            return fail(
+                f'לא הצלחתי לחלץ טקסט שמיש מ-"{file_name}" (התקבלו רק {len(body)} תווים, "{body[:60]}"). '
+                'אל תנסי לתאר או לסכם את המסמך — הטקסט לא הגיע אלייך. '
+                'הציעי למשתמש לפתוח את הקובץ ידנית או לנסות מסמך אחר.'
+            )
+        return ok(f'=== {file_name} (OCR via Vision) ===\n\n{body}')
+
     async def read_document(filePath: str) -> str:
         try:
             result = await crm.post("SmartAssistant/action/readDocument", {"filePath": filePath})
@@ -183,6 +208,127 @@ def register_document_tools(
         except Exception as e:
             return fail(f"שגיאה ביצירת מסמך מתבנית: {e}")
 
+    async def write_document_to_folder(
+        relPath: str,
+        contentBase64: str,
+    ) -> str:
+        if not case_id:
+            return fail("צריך להיות בתוך תיק כדי לכתוב לתיקייה")
+        if not relPath or not relPath.strip():
+            return fail("חסר נתיב יחסי לתיק (relPath)")
+        try:
+            result = await crm.post("SmartAssistant/action/writeToFolder", {
+                "caseId": case_id,
+                "relPath": relPath.strip(),
+                "contentBase64": contentBase64,
+            })
+            path = result.get("path")
+            size = result.get("size", 0)
+            return ok(f"✅ קובץ נכתב: {path} ({size} bytes).")
+        except Exception as e:
+            return fail(f"שגיאה בכתיבת קובץ: {e}")
+
+    async def create_subfolder(relPath: str) -> str:
+        if not case_id:
+            return fail("צריך להיות בתוך תיק כדי ליצור תיקייה")
+        if not relPath or not relPath.strip():
+            return fail("חסר נתיב לתיקייה החדשה")
+        try:
+            result = await crm.post("SmartAssistant/action/createFolder", {
+                "caseId": case_id,
+                "relPath": relPath.strip(),
+            })
+            path = result.get("path")
+            existed = result.get("alreadyExisted", False)
+            verb = "כבר קיימת" if existed else "נוצרה"
+            return ok(f"✅ תיקייה {verb}: {path}")
+        except Exception as e:
+            return fail(f"שגיאה ביצירת תיקייה: {e}")
+
+    async def rename_item(currentRelPath: str, newName: str) -> str:
+        if not case_id:
+            return fail("צריך להיות בתוך תיק כדי לשנות שם")
+        if not currentRelPath or not newName:
+            return fail("חסר currentRelPath או newName")
+        try:
+            result = await crm.post("SmartAssistant/action/renameItem", {
+                "caseId": case_id,
+                "currentRelPath": currentRelPath.strip(),
+                "newName": newName.strip(),
+            })
+            return ok(f'✅ שם שונה: "{result.get("oldPath")}" → "{result.get("newPath")}"')
+        except Exception as e:
+            return fail(f"שגיאה בשינוי שם: {e}")
+
+    async def move_item(sourceRelPath: str, targetRelPath: str) -> str:
+        if not case_id:
+            return fail("צריך להיות בתוך תיק כדי להעביר")
+        if not sourceRelPath or not targetRelPath:
+            return fail("חסר sourceRelPath או targetRelPath")
+        try:
+            result = await crm.post("SmartAssistant/action/moveItem", {
+                "caseId": case_id,
+                "sourceRelPath": sourceRelPath.strip(),
+                "targetRelPath": targetRelPath.strip(),
+            })
+            return ok(f'✅ הועבר: "{result.get("oldPath")}" → "{result.get("newPath")}"')
+        except Exception as e:
+            return fail(f"שגיאה בהעברה: {e}")
+
+    async def mark_document_for_deletion(fileRelPath: str) -> str:
+        if not case_id:
+            return fail("צריך להיות בתוך תיק כדי לסמן למחיקה")
+        if not fileRelPath:
+            return fail("חסר fileRelPath")
+        try:
+            result = await crm.post("SmartAssistant/action/markForDeletion", {
+                "caseId": case_id,
+                "fileRelPath": fileRelPath.strip(),
+            })
+            original = result.get("originalPath")
+            new = result.get("newPath")
+            return ok(
+                f'📌 המסמך סומן למחיקה (לא נמחק פיזית).\n'
+                f'הועבר ל: {new}\n'
+                f'(במקור: {original}). העו"ד יוכל למחוק ידנית מתיקיית "מסמכים למחיקה".'
+            )
+        except Exception as e:
+            return fail(f"שגיאה בסימון למחיקה: {e}")
+
+    async def create_document(
+        title: str,
+        body: str,
+        documentType: str = "letter",
+        recipient: str = "",
+    ) -> str:
+        if not case_id:
+            return fail("צריך להיות בתוך תיק כדי ליצור מסמך")
+        if not title or not title.strip():
+            return fail("חסר שם למסמך (title)")
+        if not body or not body.strip():
+            return fail("חסר תוכן למסמך (body)")
+        try:
+            result = await crm.post("SmartAssistant/action/createDocument", {
+                "caseId": case_id,
+                "title": title.strip(),
+                "body": body,
+                "documentType": documentType or "letter",
+                "recipient": recipient.strip() if recipient else None,
+            })
+            msg = result.get("message", "✅ מסמך נוצר בהצלחה")
+            doc_id = result.get("documentId")
+            net_path = result.get("networkPath")
+            extra = []
+            if doc_id:
+                extra.append(f"documentId={doc_id}")
+            if net_path:
+                extra.append(f"path={net_path}")
+            if extra:
+                msg += "\n" + " | ".join(extra)
+            return ok(msg)
+        except Exception as e:
+            return fail(f"שגיאה ביצירת מסמך: {e}")
+
     # Register
     tools["list_documents"] = {
         "description": "List documents in the case folder. Shows folder structure with file names and paths. When totalFiles is 0, returns a diagnostic explaining why (missing path, folder doesn't exist, etc.). Pass refresh=true to re-fetch from EspoCRM after the folder path changes.",
@@ -305,3 +451,112 @@ def register_document_tools(
         },
         "handler": generate_from_template,
     }
+
+    tools["write_document_to_folder"] = {
+        "description": (
+            "Write a new file to a path inside the current case's network folder. "
+            "Operations are sandboxed to the case folder — relPath is relative to the "
+            "case root (e.g. 'חוזים/2026/agreement.pdf'). The content must be base64-encoded. "
+            "Use create_document for DOCX letters; this is for raw file uploads."
+        ),
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "relPath": {"type": "string", "description": "Path inside the case folder, including filename and extension. Subfolders are created if missing."},
+                "contentBase64": {"type": "string", "description": "Base64-encoded file content."},
+            },
+            "required": ["relPath", "contentBase64"],
+        },
+        "handler": write_document_to_folder,
+    }
+
+    tools["create_subfolder"] = {
+        "description": (
+            "Create a new subfolder inside the current case's network folder. "
+            "relPath is relative to the case root. Multi-level paths are supported "
+            "(e.g. 'חוזים/2026') and missing parents are created."
+        ),
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "relPath": {"type": "string", "description": "Folder path inside the case folder."},
+            },
+            "required": ["relPath"],
+        },
+        "handler": create_subfolder,
+    }
+
+    tools["rename_item"] = {
+        "description": (
+            "Rename a file or folder inside the current case's network folder. "
+            "Does NOT move it to another folder — use move_item for that. newName "
+            "must be a plain name without slashes."
+        ),
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "currentRelPath": {"type": "string", "description": "Current path of the file/folder, relative to case root."},
+                "newName": {"type": "string", "description": "New name (without path, without slashes)."},
+            },
+            "required": ["currentRelPath", "newName"],
+        },
+        "handler": rename_item,
+    }
+
+    tools["move_item"] = {
+        "description": (
+            "Move a file or folder to a different location inside the current case's "
+            "network folder. Both source and target must stay inside the case root. "
+            "Missing intermediate folders on the target side are created."
+        ),
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "sourceRelPath": {"type": "string", "description": "Current path of the file/folder."},
+                "targetRelPath": {"type": "string", "description": "Destination path (including the final name)."},
+            },
+            "required": ["sourceRelPath", "targetRelPath"],
+        },
+        "handler": move_item,
+    }
+
+    tools["mark_document_for_deletion"] = {
+        "description": (
+            "Soft-delete a document: move it to a 'מסמכים למחיקה' subfolder inside the case "
+            "with a 'למחיקה - ' filename prefix. Does NOT physically delete. You DO NOT have "
+            "permission to delete files; always call this tool when the user asks you to "
+            "delete a document, and explain to the user that you marked it for the lawyer to "
+            "delete manually."
+        ),
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "fileRelPath": {"type": "string", "description": "Path to the file inside the case folder."},
+            },
+            "required": ["fileRelPath"],
+        },
+        "handler": mark_document_for_deletion,
+    }
+
+    tools["create_document"] = {
+        "description": (
+            "Create a new free-form DOCX document attached to the current case. "
+            "Use when no suitable template exists in availableTemplates. The body accepts "
+            "lightweight markdown: '#'/'##'/'###' for headings, '- ' for bullets, "
+            "'**bold**' and '*italic*' inline. The document is created as a Hebrew RTL DOCX "
+            "with a David font, today's date, optional recipient line, the title centered, "
+            "then the body. It is saved both as an EspoCRM Document attached to the case "
+            "and as a file in the case's network folder."
+        ),
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "title": {"type": "string", "description": "Document title shown both in the CRM record and as the centered heading."},
+                "body": {"type": "string", "description": "Document body. Lightweight markdown is supported. Plain paragraphs are fine."},
+                "documentType": {"type": "string", "enum": ["letter", "motion", "request", "general"], "description": "Document category (default 'letter')."},
+                "recipient": {"type": "string", "description": "Optional recipient line shown under the date (e.g. 'המוסד לביטוח לאומי')."},
+            },
+            "required": ["title", "body"],
+        },
+        "handler": create_document,
+    }

mcp_server/tools/legal_aid_tools.py

@@ -0,0 +1,216 @@
+"""Legal-aid billing tools: proforma → submission → monthly 320 → email payload."""
+
+from __future__ import annotations
+
+import json
+import re
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from mcp_server.espocrm_client import EspoCrmClient
+
+from mcp_server.tools._helpers import ok, fail
+
+
+MONTH_RE = re.compile(r"^\d{4}-\d{2}$")
+
+
+def _money(amount) -> str:
+    try:
+        return f"₪{float(amount):,.2f}"
+    except (TypeError, ValueError):
+        return f"₪{amount}"
+
+
+def register_legal_aid_tools(
+    tools: dict,
+    crm: "EspoCrmClient",
+    case_id: str | None,
+    user_id: str | None,
+    context: dict,
+):
+    """Register Israeli Legal-Aid (סיוע משפטי) billing tools."""
+
+    async def create_legal_aid_proforma(chargeIds: list[str]) -> str:
+        if not chargeIds:
+            return fail("צריך לפחות מזהה חיוב סיוע משפטי אחד")
+        try:
+            result = await crm.post("Invoice/action/createLegalAidProforma", {"chargeIds": chargeIds})
+            if not result.get("success"):
+                return fail(f"יצירת חשבון עסקה נכשלה: {result.get('error', 'unknown')}")
+            return ok(
+                f"חשבון עסקה (פרופורמה לסיוע משפטי) נוצר: "
+                f"{result.get('name', '')} (ID: {result.get('id')}, "
+                f"מספר חיובים: {result.get('chargesCount', len(chargeIds))}, "
+                f"סכום כולל: {_money(result.get('totalAmount'))}). "
+                f"כעת יש להגיש ידנית באתר משרד הסיוע ולאחר מכן לקרוא ל-mark_legal_aid_submitted עם מספר הבקשה שהאתר יחזיר."
+            )
+        except Exception as e:
+            return fail(f"שגיאה ביצירת פרופורמה: {e}")
+
+    async def mark_legal_aid_submitted(invoiceId: str, legalAidRequestNumber: str) -> str:
+        if not invoiceId:
+            return fail("צריך מזהה חשבון עסקה (פרופורמה)")
+        if not legalAidRequestNumber:
+            return fail("צריך מספר בקשה מאתר משרד הסיוע (לדוגמה: 2026-04-00123)")
+        try:
+            result = await crm.post(
+                f"Invoice/{invoiceId}/action/markLegalAidSubmitted",
+                {"legalAidRequestNumber": legalAidRequestNumber},
+            )
+            if not result.get("success"):
+                return fail(f"סימון הגשה נכשל: {result.get('error', 'unknown')}")
+            return ok(
+                f"פרופורמה סומנה כמוגשת. מספר בקשה: {result.get('legalAidRequestNumber')}, "
+                f"זמן הגשה: {result.get('legalAidSubmittedAt')}. "
+                f"כל החיובים המקושרים עברו לסטטוס Submitted."
+            )
+        except Exception as e:
+            return fail(f"שגיאה בסימון הגשה: {e}")
+
+    async def legal_aid_monthly_summary(month: str) -> str:
+        if not MONTH_RE.match(month or ""):
+            return fail("פורמט חודש חייב להיות YYYY-MM (לדוגמה 2026-04)")
+        try:
+            result = await crm.get(f"Invoice/action/legalAidMonthlyWrapSummary?month={month}")
+            return ok(json.dumps(result, ensure_ascii=False))
+        except Exception as e:
+            return fail(f"שגיאה בשליפת סיכום חודשי: {e}")
+
+    async def create_monthly_legal_aid_invoice(month: str) -> str:
+        if not MONTH_RE.match(month or ""):
+            return fail("פורמט חודש חייב להיות YYYY-MM (לדוגמה 2026-04)")
+        try:
+            result = await crm.post("Invoice/action/createLegalAidMonthlyTaxInvoice", {"month": month})
+            if not result.get("success"):
+                return fail(f"יצירת חשבונית חודשית נכשלה: {result.get('error', 'unknown')}")
+            return ok(
+                f"חשבונית מס חודשית (320) לחודש {month} נוצרה: "
+                f"{result.get('name', '')} (ID: {result.get('id')}, "
+                f"סכום כולל: {_money(result.get('totalAmount'))}). "
+                f"כעת ניתן להפיק payload לאימייל למשרד המשפטים באמצעות prepare_monthly_320_email."
+            )
+        except Exception as e:
+            return fail(f"שגיאה ביצירת חשבונית חודשית: {e}")
+
+    async def prepare_monthly_320_email(invoiceId: str) -> str:
+        if not invoiceId:
+            return fail("צריך מזהה חשבונית 320 חודשית")
+        try:
+            result = await crm.get(f"Invoice/{invoiceId}/action/prepareMonthly320Email")
+            payload = result.get("emailPayload") or result
+            return ok(json.dumps(payload, ensure_ascii=False))
+        except Exception as e:
+            return fail(f"שגיאה בהכנת payload לאימייל: {e}")
+
+    async def resubmit_charge(chargeId: str, newActivitySubType: str = "") -> str:
+        if not chargeId:
+            return fail("צריך מזהה חיוב")
+        body = {"id": chargeId}
+        if newActivitySubType:
+            body["newActivitySubType"] = newActivitySubType
+        try:
+            result = await crm.post("Charge/action/resubmit", body)
+            if not result.get("success"):
+                return fail("הגשה מחדש נכשלה")
+            return ok(
+                f"חיוב הוגש מחדש: ID חדש {result.get('newChargeId')} "
+                f"(מס׳ {result.get('newChargeNumber')}), "
+                f"סכום {_money(result.get('newChargeAmount'))}, "
+                f"סוג פעילות {result.get('newChargeActivitySubType')}"
+            )
+        except Exception as e:
+            return fail(f"שגיאה בהגשה מחדש: {e}")
+
+    tools["create_legal_aid_proforma"] = {
+        "description": (
+            "Create a legal-aid Proforma (Invoice type 400, חשבון עסקה) bundling Recorded charges. "
+            "All chargeIds must be legal-aid charges in 'Recorded' status. "
+            "After this, the secretary submits the proforma manually at the official Legal Aid site and gets a request number — "
+            "then call mark_legal_aid_submitted with that number."
+        ),
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "chargeIds": {"type": "array", "items": {"type": "string"}, "description": "Legal-aid charge IDs (status=Recorded)"},
+            },
+            "required": ["chargeIds"],
+        },
+        "handler": create_legal_aid_proforma,
+    }
+
+    tools["mark_legal_aid_submitted"] = {
+        "description": (
+            "Mark a legal-aid Proforma as submitted to the official Legal Aid website. "
+            "Requires the legalAidRequestNumber returned by the official site (format like '2026-04-00123'). "
+            "Flips all linked charges from Recorded to Submitted. Do NOT call without an explicit number from the user."
+        ),
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "invoiceId": {"type": "string", "description": "Proforma Invoice ID (documentType=400)"},
+                "legalAidRequestNumber": {"type": "string", "description": "Request number from official Legal Aid site"},
+            },
+            "required": ["invoiceId", "legalAidRequestNumber"],
+        },
+        "handler": mark_legal_aid_submitted,
+    }
+
+    tools["legal_aid_monthly_summary"] = {
+        "description": "Preview what a monthly 320 legal-aid tax invoice would contain for a given YYYY-MM. Read-only.",
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "month": {"type": "string", "description": "Month in YYYY-MM format (e.g. 2026-04)"},
+            },
+            "required": ["month"],
+        },
+        "handler": legal_aid_monthly_summary,
+    }
+
+    tools["create_monthly_legal_aid_invoice"] = {
+        "description": (
+            "Create the monthly type-320 tax invoice that aggregates all Paid legal-aid charges for the given YYYY-MM. "
+            "Flips all aggregated charges from Paid to Billed (terminal). "
+            "Use legal_aid_monthly_summary first to preview before creating."
+        ),
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "month": {"type": "string", "description": "Month in YYYY-MM format (e.g. 2026-04)"},
+            },
+            "required": ["month"],
+        },
+        "handler": create_monthly_legal_aid_invoice,
+    }
+
+    tools["prepare_monthly_320_email"] = {
+        "description": (
+            "Build the email payload (recipient, subject, body, attachment ref) for sending a monthly 320 legal-aid invoice "
+            "to the Justice Ministry. Returns JSON — does NOT actually send. The secretary copy-pastes or pipes via n8n."
+        ),
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "invoiceId": {"type": "string", "description": "Monthly 320 Invoice ID"},
+            },
+            "required": ["invoiceId"],
+        },
+        "handler": prepare_monthly_320_email,
+    }
+
+    tools["resubmit_charge"] = {
+        "description": (
+            "Clone a Rejected legal-aid charge as a new Recorded charge, optionally with a corrected activitySubType. "
+            "Use after the Legal Aid office rejects a line in a payment report."
+        ),
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "chargeId": {"type": "string", "description": "ID of the Rejected charge to clone"},
+                "newActivitySubType": {"type": "string", "description": "Corrected activity sub-type code (optional)"},
+            },
+            "required": ["chargeId"],
+        },
+        "handler": resubmit_charge,
+    }

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "shira-hermes"
-version = "0.2.0"
+version = "0.3.0"
 description = "Shira AI Assistant — Hermes Agent backend for EspoCRM SmartAssistant"
 requires-python = ">=3.11"
 dependencies = [
@@ -11,6 +11,7 @@ dependencies = [
     "openai>=1.50.0",
     "pyyaml>=6.0",
     "pymupdf>=1.24.0",
+    "pdfplumber>=0.11.0",
     "asyncpg>=0.30.0",
     "python-docx>=1.1.2",
     "boto3>=1.35.0",

scripts/migrate_legacy_memory_notes.py

@@ -0,0 +1,149 @@
+"""Migrate legacy "🧠 category: content" Notes into CaseMemory rows.
+
+History: until SmartAssistant 2.10.0 the Python `save_memory` tool created a
+Note entity with `post` prefixed by "🧠 {category}:" instead of using the
+proper CaseMemory entity. This left CaseMemory empty even though the system
+prompt was reading from it — so case memory never accumulated.
+
+This one-shot script reads every such Note in EspoCRM (across all cases),
+parses out the category and content, creates a corresponding CaseMemory row
+with source='assistant', and soft-deletes the original Note.
+
+Usage (from inside a shira-hermes container that already has env vars set):
+    python3 scripts/migrate_legacy_memory_notes.py [--dry-run]
+
+Safe to re-run: matches `post LIKE "🧠 %"` only; soft-deletes after migration.
+"""
+
+from __future__ import annotations
+
+import argparse
+import os
+import re
+import sys
+import time
+from typing import Any
+
+import httpx
+
+VALID_CATEGORIES = {
+    "key_facts", "strategy", "decisions",
+    "contacts_notes", "timeline",
+    "documents_notes", "billing_notes",
+}
+
+EMOJI_PREFIX_RE = re.compile(r"^🧠\s*([a-z_]+)\s*:\s*(.+)$", re.DOTALL)
+
+
+def parse_note_post(post: str) -> tuple[str, str] | None:
+    """Return (category, content) from a "🧠 category: content" string, or None."""
+    if not post or not post.startswith("🧠"):
+        return None
+    m = EMOJI_PREFIX_RE.match(post.strip())
+    if not m:
+        return None
+    category, content = m.group(1).strip(), m.group(2).strip()
+    if category not in VALID_CATEGORIES:
+        category = "key_facts"
+    return category, content
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--dry-run", action="store_true", help="Don't write anything; just report what would happen.")
+    args = parser.parse_args()
+
+    base = os.environ.get("ESPOCRM_URL", "").rstrip("/")
+    key = os.environ.get("ESPOCRM_API_KEY", "")
+    if not base or not key:
+        print("ESPOCRM_URL or ESPOCRM_API_KEY missing in env. Exiting.", file=sys.stderr)
+        return 2
+
+    headers = {"X-Api-Key": key, "Content-Type": "application/json"}
+
+    # Page through notes that look like memory entries.
+    offset = 0
+    page_size = 100
+    migrated = 0
+    skipped = 0
+    failed = 0
+
+    while True:
+        params = {
+            "where[0][type]": "startsWith",
+            "where[0][attribute]": "post",
+            "where[0][value]": "🧠",
+            "maxSize": page_size,
+            "offset": offset,
+            "select": "id,post,parentId,parentType,createdAt,createdById",
+        }
+        r = httpx.get(f"{base}/api/v1/Note", headers=headers, params=params, timeout=30)
+        if r.status_code != 200:
+            print(f"GET /Note failed: HTTP {r.status_code} — {r.text[:200]}", file=sys.stderr)
+            return 1
+
+        data = r.json()
+        rows: list[dict[str, Any]] = data.get("list", []) or []
+        if not rows:
+            break
+
+        for row in rows:
+            parent_type = row.get("parentType")
+            parent_id = row.get("parentId")
+            post = row.get("post") or ""
+            note_id = row.get("id")
+
+            if parent_type != "Case" or not parent_id:
+                skipped += 1
+                continue
+
+            parsed = parse_note_post(post)
+            if not parsed:
+                skipped += 1
+                continue
+
+            category, content = parsed
+            name = content[:100]
+            print(f"  migrate Note {note_id} [{parent_id}] → CaseMemory ({category}): {name[:60]}")
+
+            if args.dry_run:
+                migrated += 1
+                continue
+
+            create_body = {
+                "name": name,
+                "caseId": parent_id,
+                "category": category,
+                "content": content,
+                "source": "assistant",
+                "importance": "normal",
+            }
+            cr = httpx.post(f"{base}/api/v1/CaseMemory", headers=headers, json=create_body, timeout=30)
+            if cr.status_code not in (200, 201):
+                print(f"    CREATE CaseMemory failed: HTTP {cr.status_code} — {cr.text[:200]}", file=sys.stderr)
+                failed += 1
+                continue
+
+            dr = httpx.delete(f"{base}/api/v1/Note/{note_id}", headers=headers, timeout=30)
+            if dr.status_code not in (200, 204):
+                print(f"    DELETE Note {note_id} failed: HTTP {dr.status_code} — {dr.text[:200]}", file=sys.stderr)
+            migrated += 1
+
+        offset += page_size
+        # Don't loop forever on a misbehaving total.
+        total = data.get("total", 0)
+        if offset >= total:
+            break
+        time.sleep(0.2)  # gentle rate-limit
+
+    print()
+    print(f"=== summary ===")
+    print(f"  migrated: {migrated}")
+    print(f"  skipped:  {skipped}")
+    print(f"  failed:   {failed}")
+    print(f"  mode:     {'dry-run' if args.dry_run else 'live'}")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())