feat(tools): add Twenty CRM sync scripts and update .gitignore rules

tools/gather_recent_context.py

@@ -0,0 +1,211 @@
+#!/usr/bin/env python3
+"""Sync job-opportunity email context into Twenty CRM as Notes.
+
+Called from the /gather-recent-context command. The command owns the email
+side (Fastmail MCP: search_email / read_thread) and writes the classified
+threads to a JSON file; this script owns everything deterministic:
+
+  1. Match each email thread to an opportunity on an ACTIVE loop
+     (stage not in the final set) by normalized company name, breaking ties
+     with role-title token overlap against the opportunity name.
+  2. Build one "Email Context — <Company>" note per matched opportunity with
+     a dated markdown timeline of the thread summaries.
+  3. Dedup by thread id against job_scraper/context_sync.json so a re-run
+     never writes the same conversation twice.
+  4. Apply via Twenty REST (TWENTY_API_KEY) with --apply, or emit
+     job_scraper/context_plan.json for the command to apply through the
+     Twenty MCP session.
+
+Input emails JSON (written by the command):
+  {"threads": [{"thread_id": "...", "company": "Acme", "role": "Backend Engineer"|null,
+                "date": "YYYY-MM-DD", "from": "recruiter@acme.com",
+                "subject": "...", "summary": "1-3 sentence summary",
+                "signal": "interview|assessment|offer|rejection|scheduling|other"}]}
+
+Usage:
+  python3 tools/gather_recent_context.py --emails <emails.json>            # plan only
+  python3 tools/gather_recent_context.py --emails <emails.json> --apply   # plan + REST apply
+  ... --snapshot crm.json    # offline CRM state instead of REST fetch
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from datetime import date
+from pathlib import Path
+
+from twenty_enrich import norm_company, norm_title
+
+REPO = Path(__file__).resolve().parent.parent
+STATE = REPO / "job_scraper" / "context_sync.json"
+PLAN_OUT = REPO / "job_scraper" / "context_plan.json"
+
+# Loops still in play: everything before a terminal outcome.
+ACTIVE_STAGES = {"RANKED", "DRAFTED", "APPLIED", "INTERVIEW", "OFFER"}
+
+
+def load_state(path: Path = STATE) -> dict:
+    if path.exists():
+        return json.loads(path.read_text(encoding="utf-8"))
+    return {"processed_thread_ids": [], "last_run": None}
+
+
+def match_thread(thread: dict, opportunities: list[dict],
+                 company_name_by_id: dict[str, str]) -> dict | None:
+    """Best active-loop opportunity for a thread, or None.
+
+    Opportunity names are role-only (the ' — Company' suffix is stripped once
+    the company relation is set - see tools/twenty_enrich.py), so the company
+    match goes through ``companyId`` -> Company.name. Among a company's active
+    opportunities, prefer the one whose role name shares the most title tokens
+    with the thread's role (when the thread names one); a company with exactly
+    one active loop matches unconditionally.
+    """
+    key = norm_company(thread.get("company", ""))
+    if not key:
+        return None
+    candidates = [o for o in opportunities
+                  if o.get("stage") in ACTIVE_STAGES
+                  and norm_company(company_name_by_id.get(o.get("companyId") or "", "")) == key]
+    if not candidates:
+        return None
+    if len(candidates) == 1 or not thread.get("role"):
+        return candidates[0]
+    role_tokens = set(norm_title(thread["role"]).split())
+
+    def overlap(o):
+        return len(role_tokens & set(norm_title(o.get("name", "")).split()))
+
+    return max(candidates, key=overlap)
+
+
+def build_note_markdown(threads: list[dict]) -> str:
+    """Dated timeline, newest first. Only classified thread data - no bodies."""
+    lines = []
+    for t in sorted(threads, key=lambda t: t.get("date") or "", reverse=True):
+        sig = t.get("signal", "other")
+        lines.append(f"## {t.get('date', 'undated')} — {sig}")
+        lines.append(f"**From:** {t.get('from', 'unknown')} · **Subject:** {t.get('subject', '')}")
+        lines.append("")
+        lines.append(t.get("summary", "").strip())
+        lines.append("")
+    return "\n".join(lines).strip()
+
+
+def build_plan(threads: list[dict], opportunities: list[dict],
+               companies: list[dict], state: dict) -> dict:
+    """Notes plan grouped one note per opportunity, dedup'd by thread id."""
+    company_name_by_id = {c["id"]: c.get("name", "") for c in companies}
+    seen_ids = set(state.get("processed_thread_ids", []))
+    fresh = [t for t in threads if t.get("thread_id") and t["thread_id"] not in seen_ids]
+    by_opp: dict[str, dict] = {}
+    unmatched = []
+    for t in fresh:
+        o = match_thread(t, opportunities, company_name_by_id)
+        if o is None:
+            unmatched.append({"thread_id": t["thread_id"],
+                              "company": t.get("company"), "subject": t.get("subject")})
+            continue
+        slot = by_opp.setdefault(o["id"], {"opportunity_id": o["id"],
+                                           "opportunity_name": o.get("name"),
+                                           "company": company_name_by_id.get(o.get("companyId") or "", ""),
+                                           "threads": []})
+        slot["threads"].append(t)
+
+    notes = []
+    for slot in by_opp.values():
+        company = slot["company"] or "Unknown"
+        notes.append({
+            "opportunity_id": slot["opportunity_id"],
+            "title": f"Email Context — {company} ({date.today().isoformat()})",
+            "markdown": build_note_markdown(slot["threads"]),
+            "thread_ids": [t["thread_id"] for t in slot["threads"]],
+        })
+    return {"notes": notes, "unmatched_threads": unmatched,
+            "skipped_already_processed": len(threads) - len(fresh)}
+
+
+def apply_plan(plan: dict, client) -> None:
+    """Create the notes + note targets via REST, then update dedup state."""
+    for n in plan["notes"]:
+        created = client.create_batch("notes", [{
+            "title": n["title"],
+            "bodyV2": {"markdown": n["markdown"]},
+            "position": "last",
+        }])
+        note_id = created[0]["id"]
+        client.create_batch("noteTargets", [{
+            "noteId": note_id,
+            "targetOpportunityId": n["opportunity_id"],
+            "position": "last",
+        }])
+        n["note_id"] = note_id
+
+
+def commit_state(plan: dict, state: dict, path: Path = STATE) -> None:
+    done = set(state.get("processed_thread_ids", []))
+    for n in plan["notes"]:
+        done.update(n["thread_ids"])
+    state["processed_thread_ids"] = sorted(done)
+    state["last_run"] = date.today().isoformat()
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(state, indent=2), encoding="utf-8")
+
+
+def main(argv=None) -> int:
+    ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    ap.add_argument("--emails", type=Path,
+                    help="classified threads JSON written by /gather-recent-context")
+    ap.add_argument("--snapshot", type=Path,
+                    help="CRM opportunities JSON (offline; else fetched via REST)")
+    ap.add_argument("--apply", action="store_true",
+                    help="create the notes via Twenty REST (needs TWENTY_API_KEY)")
+    ap.add_argument("--commit-plan", type=Path,
+                    help="mark an already-applied plan's thread ids as processed and exit "
+                         "(used after the command applies the plan through the Twenty MCP session)")
+    ap.add_argument("--out", type=Path, default=PLAN_OUT, help="plan output path")
+    args = ap.parse_args(argv)
+
+    if args.commit_plan:
+        plan = json.loads(args.commit_plan.read_text(encoding="utf-8"))
+        commit_state(plan, load_state())
+        print(f"state committed for {len(plan['notes'])} notes.")
+        return 0
+    if not args.emails:
+        ap.error("--emails is required unless --commit-plan is given")
+
+    threads = json.loads(args.emails.read_text(encoding="utf-8"))["threads"]
+    client = None
+    if args.snapshot:
+        snap = json.loads(args.snapshot.read_text(encoding="utf-8"))
+        opportunities, companies = snap["opportunities"], snap["companies"]
+    else:
+        from twenty_client import TwentyClient
+        client = TwentyClient()
+        opportunities = client.list_all("opportunities")
+        companies = client.list_all("companies")
+
+    state = load_state()
+    plan = build_plan(threads, opportunities, companies, state)
+
+    if args.apply and plan["notes"]:
+        if client is None:
+            from twenty_client import TwentyClient
+            client = TwentyClient()
+        apply_plan(plan, client)
+        commit_state(plan, state)
+        plan["applied"] = True
+
+    args.out.parent.mkdir(parents=True, exist_ok=True)
+    args.out.write_text(json.dumps(plan, indent=2), encoding="utf-8")
+    print(f"plan -> {args.out}")
+    print(f"notes: {len(plan['notes'])} · unmatched threads: {len(plan['unmatched_threads'])}"
+          f" · already processed: {plan['skipped_already_processed']}")
+    if args.apply:
+        print("applied via REST; state updated." if plan["notes"] else "nothing to apply.")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

tools/security_guards.py

@@ -45,13 +45,18 @@ ALLOWED_PERMISSIONS = {
 }
 
 # Personal-data ignore rules that must never disappear from .gitignore.
+# Fork policy: documents/ (cv, linkedin, diplomas, references, postings,
+# sources) is deliberately tracked in git - this repo is a personal fork,
+# not the upstream template. Only generated/derived personal state and
+# secrets stay ignored.
 REQUIRED_IGNORE_RULES = [
     "salary_data.json",
     # Depth-independent: the job-scraper skill resolves `job_scraper/` relative
     # to its own directory, so the state file lands under .claude/skills/... and
     # a repo-rooted rule silently fails to match it.
     "**/job_scraper/seen_jobs.json",
-    "**/job_scraper/notion_sync.json",
+    "**/job_scraper/twenty_sync.json",
+    "**/job_scraper/context_sync.json",
     "**/job_scraper/*.md",
     "*_BehavioralReport.pdf",
     "linkedin_Profile.pdf",
@@ -62,15 +67,9 @@ REQUIRED_IGNORE_RULES = [
     "cover_letters/cover_*.*",
     # /apply also recognizes the uppercase Cover_* naming variant.
     "cover_letters/Cover_*.*",
-    "documents/cv/**",
-    "documents/linkedin/**",
-    "documents/diplomas/**",
-    "documents/references/**",
     "documents/applications/**",
-    "documents/postings/**",
     "documents/interview/**",
     "job_search_tracker.csv",
-    "gmail_sync/",
     "reports/",
     "upskill/*.md",
     # Depth-independent twin of the rule above. The upskill *skill* resolves

tools/twenty_client.py

@@ -0,0 +1,100 @@
+"""Minimal Twenty CRM REST client (stdlib only).
+
+Shared by tools/twenty_enrich.py and tools/gather_recent_context.py.
+
+Auth: set TWENTY_API_KEY (workspace API key from Settings -> API & Webhooks)
+and optionally TWENTY_BASE_URL (default: https://crm.prad.nu).
+
+Both consumer scripts also run keyless: they compute a *plan* JSON from local
+data plus an optional CRM snapshot, and the /sync-twenty or
+/gather-recent-context command applies the plan through the Twenty MCP
+server, which is session-authenticated via OAuth. The REST path exists so the
+whole loop can also run headless (cron, CI) with an API key.
+"""
+from __future__ import annotations
+
+import json
+import os
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
+
+DEFAULT_BASE_URL = "https://crm.prad.nu"
+PAGE_LIMIT = 60  # Twenty REST caps page size at 60
+
+
+class TwentyError(RuntimeError):
+    pass
+
+
+class TwentyClient:
+    def __init__(self, base_url: str | None = None, api_key: str | None = None):
+        self.base_url = (base_url or os.environ.get("TWENTY_BASE_URL") or DEFAULT_BASE_URL).rstrip("/")
+        self.api_key = api_key or os.environ.get("TWENTY_API_KEY")
+        if not self.api_key:
+            raise TwentyError(
+                "TWENTY_API_KEY is not set. Either export it, or run without --apply "
+                "to emit a plan JSON and apply it through the Twenty MCP session."
+            )
+
+    def _request(self, method: str, path: str, body: dict | None = None,
+                 params: dict | None = None, retries: int = 3):
+        url = f"{self.base_url}{path}"
+        if params:
+            url += "?" + urllib.parse.urlencode(params)
+        data = json.dumps(body).encode() if body is not None else None
+        last_err = None
+        for attempt in range(retries):
+            req = urllib.request.Request(url, data=data, method=method, headers={
+                "Authorization": f"Bearer {self.api_key}",
+                "Content-Type": "application/json",
+                "Accept": "application/json",
+            })
+            try:
+                with urllib.request.urlopen(req, timeout=60) as resp:
+                    return json.loads(resp.read().decode())
+            except urllib.error.HTTPError as e:  # 4xx never retried: same input, same answer
+                detail = e.read().decode(errors="replace")[:500]
+                if e.code == 429 or e.code >= 500:
+                    last_err = TwentyError(f"{method} {path} -> {e.code}: {detail}")
+                    time.sleep(1.5 * (attempt + 1))
+                    continue
+                raise TwentyError(f"{method} {path} -> {e.code}: {detail}") from e
+            except urllib.error.URLError as e:
+                last_err = TwentyError(f"{method} {path} -> {e.reason}")
+                time.sleep(1.5 * (attempt + 1))
+        raise last_err
+
+    # -- records ------------------------------------------------------------
+
+    def list_all(self, plural: str, order_by: str = "id") -> list[dict]:
+        """Every record of an object, cursor-paged."""
+        out: list[dict] = []
+        cursor = None
+        while True:
+            params = {"limit": PAGE_LIMIT, "order_by": order_by}
+            if cursor:
+                params["starting_after"] = cursor
+            payload = self._request("GET", f"/rest/{plural}", params=params)
+            page = payload.get("data", {}).get(plural, [])
+            out.extend(page)
+            page_info = payload.get("pageInfo", {})
+            if not page_info.get("hasNextPage") or not page:
+                break
+            cursor = page_info.get("endCursor") or page[-1]["id"]
+        return out
+
+    def create_batch(self, plural: str, records: list[dict]) -> list[dict]:
+        """Create records in chunks of PAGE_LIMIT via the batch endpoint."""
+        created: list[dict] = []
+        for i in range(0, len(records), PAGE_LIMIT):
+            payload = self._request("POST", f"/rest/batch/{plural}",
+                                    body=records[i:i + PAGE_LIMIT])
+            created.extend(payload.get("data", {}).get(plural, []))
+        return created
+
+    def update_one(self, plural: str, record_id: str, patch: dict) -> dict:
+        payload = self._request("PATCH", f"/rest/{plural}/{record_id}", body=patch)
+        data = payload.get("data", {})
+        return next(iter(data.values())) if data else payload

tools/twenty_enrich.py

@@ -0,0 +1,295 @@
+#!/usr/bin/env python3
+"""Enrich Twenty CRM opportunities and companies from local job-search data.
+
+Encodes the deterministic layers of the /sync-twenty enrichment pipeline:
+
+  Opportunity salary (``amount``), in source priority order:
+    1. seen_jobs.json ``salary_max_stated``   (stated in the posting at rank time)
+    2. jobs.db postings salary_max            (same company+title, stated on an ATS board)
+    3. jobs.db companies levels_swe_median    (levels.fyi company median - labeled estimate)
+
+  Company fields (``domainName``, ``linkedinLink``, ``address``, ``annualRevenue``):
+    - jobs.db companies row matched on normalized name, homepage domain joined
+      through postings.company_domain. Stringified NaN / empty strings are absent.
+
+Anything these layers cannot ground is emitted in the plan's ``unresolved``
+section for the calling command to escalate to web research (scout subagents /
+Perplexity) - this script never guesses.
+
+Usage:
+  python3 tools/twenty_enrich.py                 # plan -> job_scraper/enrichment_plan.json
+  python3 tools/twenty_enrich.py --apply         # plan + apply via REST (needs TWENTY_API_KEY)
+  python3 tools/twenty_enrich.py --snapshot f.json   # offline: CRM state from a JSON snapshot
+
+Snapshot shape (what /sync-twenty exports through MCP):
+  {"opportunities": [{id, url:{primaryLinkUrl}, amount, closeDate, ...}],
+   "companies":     [{id, name, domainName, linkedinLink, address, ...}]}
+
+The plan JSON is idempotent to apply: every entry names the target record id
+and only fields that are currently empty.
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import re
+import sqlite3
+import sys
+from pathlib import Path
+
+REPO = Path(__file__).resolve().parent.parent
+SEEN_JOBS = REPO / "job_scraper" / "seen_jobs.json"
+JOBS_DB = REPO / "documents" / "postings" / "jobs.db"
+PLAN_OUT = REPO / "job_scraper" / "enrichment_plan.json"
+
+_SUFFIXES = re.compile(r"\b(inc|llc|ltd|corp|corporation|co|labs|lab|hq|the)\b")
+
+
+def norm_company(s: str) -> str:
+    """Normalized company key: lowercase, punctuation collapsed, suffixes stripped."""
+    s = re.sub(r"[^a-z0-9]+", " ", (s or "").lower()).strip()
+    s = _SUFFIXES.sub("", s).strip()
+    return re.sub(r"\s+", " ", s)
+
+
+def norm_title(s: str) -> str:
+    return re.sub(r"[^a-z0-9]+", " ", (s or "").lower()).strip()
+
+
+def clean_domain(raw: str | None) -> str | None:
+    """Bare registrable domain or None. Rejects the stringified-NaN artifact."""
+    if not raw:
+        return None
+    d = str(raw).strip().lower()
+    for prefix in ("http://", "https://", "www."):
+        d = d.removeprefix(prefix)
+    d = d.rstrip("/")
+    if not d or d == "nan" or "." not in d:
+        return None
+    return d
+
+
+def _has_link(v) -> bool:
+    return bool(v and v.get("primaryLinkUrl"))
+
+
+def _has_addr(v) -> bool:
+    return bool(v and (v.get("addressCity") or v.get("addressCountry")))
+
+
+def _has_amount(v) -> bool:
+    return bool(v and v.get("amountMicros"))
+
+
+def load_local_sources(seen_path: Path = SEEN_JOBS, db_path: Path = JOBS_DB):
+    """Ranked jobs plus the jobs.db lookup tables the plan builders consume."""
+    seen = json.loads(seen_path.read_text(encoding="utf-8"))["seen"]
+    ranked = [v for v in seen.values() if v.get("status") == "ranked"]
+
+    db_companies: dict[str, dict] = {}
+    domains: dict[str, str] = {}
+    postings_salary: dict[tuple[str, str], int] = {}
+    levels: dict[str, int] = {}
+    if db_path.exists():
+        con = sqlite3.connect(db_path)
+        con.row_factory = sqlite3.Row
+        for r in con.execute(
+            "SELECT company_id, company_name, linkedin_url, country, state, apollo_city,"
+            " organization_revenue, levels_swe_median_usd, levels_median_comp_usd FROM companies"
+        ):
+            key = norm_company(r["company_name"])
+            db_companies.setdefault(key, dict(r))
+            lv = r["levels_swe_median_usd"] or r["levels_median_comp_usd"]
+            if lv and lv > 0:
+                levels.setdefault(key, int(lv))
+        for r in con.execute(
+            "SELECT company_id, company_domain, COUNT(*) n FROM postings"
+            " WHERE company_domain IS NOT NULL AND company_domain != ''"
+            " GROUP BY company_id, company_domain ORDER BY n"
+        ):
+            domains[r["company_id"]] = r["company_domain"]  # last row = most frequent
+        for r in con.execute(
+            "SELECT company_name, title, salary_max, salary_currency FROM postings"
+            " WHERE salary_max IS NOT NULL AND salary_max > 0"
+        ):
+            if (r["salary_currency"] or "USD").upper() in ("USD", "$", ""):
+                key = (norm_company(r["company_name"]), norm_title(r["title"]))
+                postings_salary.setdefault(key, int(float(r["salary_max"])))
+        con.close()
+    return ranked, db_companies, domains, postings_salary, levels
+
+
+def build_salary_plan(ranked, opportunities, postings_salary, levels):
+    """(updates, unresolved): amount patches for opportunities missing one."""
+    by_url = {}
+    for o in opportunities:
+        u = (o.get("url") or {}).get("primaryLinkUrl")
+        if u:
+            by_url[u] = o
+    updates, unresolved = [], []
+    for v in ranked:
+        o = by_url.get(v.get("url"))
+        if not o or _has_amount(o.get("amount")):
+            continue
+        amount = source = None
+        if v.get("salary_max_stated"):
+            amount, source = int(v["salary_max_stated"]), "seen_jobs.salary_max_stated"
+        else:
+            key = (norm_company(v.get("company", "")), norm_title(v.get("title", "")))
+            if key in postings_salary:
+                amount, source = postings_salary[key], "jobs.db postings"
+            elif norm_company(v.get("company", "")) in levels:
+                amount = levels[norm_company(v.get("company", ""))]
+                source = "levels.fyi company SWE median (estimate)"
+        if amount:
+            updates.append({
+                "id": o["id"],
+                "source": source,
+                "patch": {"amount": {"amountMicros": amount * 1_000_000, "currencyCode": "USD"}},
+            })
+        else:
+            unresolved.append({"id": o["id"], "url": v["url"],
+                               "company": v.get("company"), "title": v.get("title")})
+    return updates, unresolved
+
+
+def build_company_plan(companies, db_companies, domains):
+    """(updates, unresolved): fill ONLY empty company fields from jobs.db."""
+    updates, unresolved = [], []
+    for c in companies:
+        key = norm_company(c.get("name", ""))
+        m = db_companies.get(key) or {}
+        patch = {}
+        if not _has_link(c.get("domainName")):
+            dom = clean_domain(domains.get(m.get("company_id", "")))
+            if dom:
+                patch["domainName"] = {"primaryLinkUrl": f"https://{dom}",
+                                       "primaryLinkLabel": c["name"]}
+        if not _has_link(c.get("linkedinLink")):
+            li = (m.get("linkedin_url") or "").strip()
+            if li:
+                patch["linkedinLink"] = {"primaryLinkUrl": li.replace("http://", "https://"),
+                                         "primaryLinkLabel": c["name"]}
+        if not _has_addr(c.get("address")):
+            addr = {k: v for k, v in [("addressCity", m.get("apollo_city")),
+                                      ("addressState", m.get("state")),
+                                      ("addressCountry", m.get("country"))] if v}
+            if addr:
+                patch["address"] = addr
+        if patch:
+            updates.append({"id": c["id"], "name": c["name"], "patch": patch})
+        # Fields still empty after the local layer -> web-research escalation
+        missing = []
+        if not _has_link(c.get("domainName")) and "domainName" not in patch:
+            missing.append("domain")
+        if not _has_link(c.get("linkedinLink")) and "linkedinLink" not in patch:
+            missing.append("linkedin")
+        if not _has_addr(c.get("address")) and "address" not in patch:
+            missing.append("address")
+        if missing:
+            unresolved.append({"id": c["id"], "name": c["name"], "missing": missing})
+    return updates, unresolved
+
+
+def build_name_cleanup_plan(opportunities):
+    """Strip the ' — <Company>' suffix from opportunity names.
+
+    Runs LAST, and only for opportunities whose company relation is already
+    set: the suffix is the human-readable fallback while the record is
+    unlinked, and becomes redundant once ``companyId`` points at the Company.
+    An unlinked opportunity keeps its suffixed name so it stays identifiable.
+    """
+    updates = []
+    for o in opportunities:
+        name = o.get("name") or ""
+        if o.get("companyId") and " — " in name:
+            role = name.split(" — ")[0].strip()
+            if role and role != name:
+                updates.append({"id": o["id"], "patch": {"name": role}})
+    return updates
+
+
+def coverage(opportunities, companies) -> dict:
+    n, m = len(companies), len(opportunities)
+    return {
+        "companies": {
+            "total": n,
+            "domain": sum(1 for c in companies if _has_link(c.get("domainName"))),
+            "linkedin": sum(1 for c in companies if _has_link(c.get("linkedinLink"))),
+            "address": sum(1 for c in companies if _has_addr(c.get("address"))),
+        },
+        "opportunities": {
+            "total": m,
+            "salary": sum(1 for o in opportunities if _has_amount(o.get("amount"))),
+            "company_link": sum(1 for o in opportunities if o.get("companyId")),
+            "score": sum(1 for o in opportunities if o.get("score") is not None),
+        },
+    }
+
+
+def main(argv=None) -> int:
+    ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    ap.add_argument("--snapshot", type=Path,
+                    help="CRM state JSON (offline mode; else fetched via REST)")
+    ap.add_argument("--apply", action="store_true",
+                    help="apply the plan via Twenty REST (needs TWENTY_API_KEY)")
+    ap.add_argument("--out", type=Path, default=PLAN_OUT, help="plan output path")
+    args = ap.parse_args(argv)
+
+    ranked, db_companies, domains, postings_salary, levels = load_local_sources()
+
+    client = None
+    if args.snapshot:
+        snap = json.loads(args.snapshot.read_text(encoding="utf-8"))
+        opportunities, companies = snap["opportunities"], snap["companies"]
+    else:
+        from twenty_client import TwentyClient  # noqa: deferred import keeps offline mode key-free
+        client = TwentyClient()
+        opportunities = client.list_all("opportunities")
+        companies = client.list_all("companies")
+
+    sal_updates, sal_unresolved = build_salary_plan(ranked, opportunities, postings_salary, levels)
+    comp_updates, comp_unresolved = build_company_plan(companies, db_companies, domains)
+    name_updates = build_name_cleanup_plan(opportunities)
+
+    plan = {
+        "coverage_before": coverage(opportunities, companies),
+        "opportunity_updates": sal_updates,
+        "company_updates": comp_updates,
+        "name_cleanup": name_updates,
+        "unresolved": {
+            "opportunities_without_salary": sal_unresolved,
+            "companies_missing_fields": comp_unresolved,
+        },
+    }
+
+    if args.apply:
+        if client is None:
+            from twenty_client import TwentyClient
+            client = TwentyClient()
+        for u in sal_updates:
+            client.update_one("opportunities", u["id"], u["patch"])
+        for u in comp_updates:
+            client.update_one("companies", u["id"], u["patch"])
+        # Name cleanup LAST: it must observe the post-link state.
+        for u in name_updates:
+            client.update_one("opportunities", u["id"], u["patch"])
+        plan["applied"] = {"opportunities": len(sal_updates), "companies": len(comp_updates),
+                           "names_cleaned": len(name_updates)}
+
+    args.out.parent.mkdir(parents=True, exist_ok=True)
+    args.out.write_text(json.dumps(plan, indent=2), encoding="utf-8")
+    cov = plan["coverage_before"]
+    print(f"plan -> {args.out}")
+    print(f"opportunity salary updates: {len(sal_updates)} "
+          f"(coverage {cov['opportunities']['salary']}/{cov['opportunities']['total']} before)")
+    print(f"company field updates: {len(comp_updates)} · name cleanups: {len(name_updates)}")
+    print(f"unresolved -> web research: {len(sal_unresolved)} postings, "
+          f"{len(comp_unresolved)} companies")
+    if args.apply:
+        print("applied via REST.")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())