Adopt the Pipeline: grid-driven opportunities link (v0.1.0:87)

The fundraising grid (canonical) now drives the classic opportunities
Pipeline board, instead of the board being a disconnected second data-entry
surface. An "Add to Pipeline" row action creates a durably-linked opportunity
via the new opportunities.fundraising_investor_id (migration 0005, additive +
reversible), reusing the grid's already-synced contact — retiring the
POST /api/contacts side-door — and mapping the grid lead to the opp owner.

Ownership is split so the two stay reconciled: the grid owns whether the link
exists and the seed; the board owns stage/probability/owner. The link endpoint
is idempotent (one live opp per investor; a re-link never reseeds funnel
fields). "Is in pipeline?"/"what stage?" are derived from a live opp join and
injected as read-only grid columns on read, stripped on write, so they never
persist or dirty the autosave. Remove-from-pipeline soft-deletes the opp and
leaves the grid row fully intact; deleting an investor from the grid archives
its orphaned opp.

Also fixes the standing soft-delete leak in handle_pipeline_report and the
dashboard pipeline aggregates, which counted tombstoned opportunities.

Tests: backend/test_grid_pipeline_link.py (link/idempotent/round-trip/guards/
unlink-intact/re-link/orphan/aggregates); 28/28 suite green, render-smoke green.

backend/migrations/0005_grid_pipeline_link.down.sql

@@ -0,0 +1,7 @@
+-- Reversal of 0005_grid_pipeline_link.sql (manual; .down files are never auto-applied).
+--
+-- SQLite < 3.35 cannot DROP COLUMN. The added column is nullable and ignored by any code
+-- path predating it, so leaving it in place is harmless. The index drops freely. On
+-- SQLite >= 3.35 the column itself may also be dropped.
+DROP INDEX IF EXISTS idx_opportunities_fr_investor;
+-- ALTER TABLE opportunities DROP COLUMN fundraising_investor_id;  -- SQLite >= 3.35 only

backend/migrations/0005_grid_pipeline_link.sql

@@ -0,0 +1,22 @@
+-- Grid → Pipeline adoption — a durable link from a fundraising-grid investor to its
+-- Pipeline opportunity row.
+--
+-- ADDITIVE + REVERSIBLE (CLAUDE.md guardrail #3): adds one nullable column + index.
+-- Until now the grid's "Create Opportunity" button fired a one-shot POST with no
+-- back-reference, so a grid investor could spawn unlimited duplicate opportunities and
+-- an opp never knew which grid row it belonged to. opportunities.fundraising_investor_id
+-- records the link (set by the new POST /api/fundraising/pipeline/link endpoint), making
+-- the relationship dedup-able and reconcilable. "Is this investor in the pipeline?" and
+-- "what stage?" are then DERIVED from a live join on this column — deliberately not a
+-- denormalized mirror flag on fundraising_investors, which would only reintroduce the
+-- two-model drift this CRM exists to fight.
+--
+-- fundraising_investor_id is a LOGICAL foreign key to fundraising_investors(id). It is
+-- intentionally NOT a declared SQLite FOREIGN KEY: opportunities are soft-deleted (never
+-- hard-deleted) and fundraising_investors rows are rebuilt on every grid save, so there
+-- is nothing to cascade; SQLite's ALTER TABLE ADD COLUMN cannot add an enforced FK
+-- cleanly anyway. Nullable so every existing opportunity stays valid — a manually-created,
+-- non-grid opportunity simply has NULL here.
+ALTER TABLE opportunities ADD COLUMN fundraising_investor_id TEXT;
+
+CREATE INDEX IF NOT EXISTS idx_opportunities_fr_investor ON opportunities(fundraising_investor_id);

backend/server.py

@@ -1584,6 +1584,12 @@ def sanitize_fundraising_grid(grid):
     if not isinstance(rows, list):
         rows = deep_copy_json(DEFAULT_FUNDRAISING_ROWS)
 
+    # `pipeline` / `pipeline_stage` are read-only columns whose VALUES are derived from the
+    # linked opportunity and injected on read — never persisted as row data (the GET handler
+    # re-injects them after sanitize). The column DEFINITIONS persist like any other column
+    # so their position / width / hidden state is kept.
+    _computed_row_values = ('longshot_followup', 'pipeline', 'pipeline_stage')
+
     clean_columns = []
     seen = set()
     for col in columns:
@@ -1600,11 +1606,83 @@ def sanitize_fundraising_grid(grid):
         if not isinstance(row, dict):
             continue
         next_row = dict(row)
-        next_row.pop('longshot_followup', None)
+        for _k in _computed_row_values:
+            next_row.pop(_k, None)
         clean_rows.append(next_row)
 
     return {"columns": clean_columns, "rows": clean_rows}
 
+# ─── Grid ↔ Pipeline link (Adopt the Pipeline) ────────────────────────────────
+# The fundraising grid is canonical; the Pipeline board is a view of the deals it
+# drives. opportunities.fundraising_investor_id is the durable join. Two ownership
+# rules keep them reconciled:
+#   * Grid owns: whether the link exists, the investor, the primary contact, the seed.
+#   * Pipeline owns: stage / probability / owner / close date / next step.
+# So a grid save NEVER reseeds an existing linked opp (it would clobber funnel state),
+# and "is in pipeline?" / "what stage?" are DERIVED from the live opp join — never a
+# denormalized flag that could drift.
+
+def _resolve_owner_from_lead(conn, lead_value, fallback_user_id):
+    """Map a grid 'lead' cell (a team member's name/initials) to a users.id.
+    opportunities.owner_id is NOT NULL, so a value is always returned — the acting user
+    is the fallback when no confident match exists. Owner is reassignable on the board,
+    so a forgiving prefix match is acceptable here."""
+    lead = str(lead_value or '').strip().lower()
+    if lead:
+        row = conn.execute(
+            "SELECT id FROM users WHERE is_active = 1 AND (lower(full_name) = ? OR lower(username) = ?) LIMIT 1",
+            (lead, lead)
+        ).fetchone()
+        if row:
+            return row['id']
+        row = conn.execute(
+            "SELECT id FROM users WHERE is_active = 1 AND (lower(full_name) LIKE ? OR lower(username) LIKE ?) "
+            "ORDER BY length(full_name) LIMIT 1",
+            (lead + '%', lead + '%')
+        ).fetchone()
+        if row:
+            return row['id']
+    return fallback_user_id
+
+
+def reconcile_grid_pipeline_links(conn):
+    """After a grid save + relational sync, archive (soft-delete) any pipeline
+    opportunity whose linked grid investor row no longer exists — i.e. the investor was
+    deleted from the grid. Creation is NEVER done here: a pipeline opp is only created
+    via the explicit /api/fundraising/pipeline/link endpoint (which carries the seed
+    fields), so this reconciler is a one-way orphan cleanup that can never spawn an
+    empty opp or reseed a live one."""
+    conn.execute(
+        """
+        UPDATE opportunities
+        SET deleted_at = ?, updated_at = ?
+        WHERE fundraising_investor_id IS NOT NULL
+          AND deleted_at IS NULL
+          AND fundraising_investor_id NOT IN (SELECT id FROM fundraising_investors)
+        """,
+        (now(), now())
+    )
+
+
+def pipeline_stage_by_source_row(conn):
+    """Return {grid source_row_id: current pipeline stage} for every investor with a
+    live (non-deleted) linked opportunity. The opportunities table is the single source
+    of truth, so this is always derived fresh and injected as read-only grid columns —
+    never stored in the grid blob, where it could go stale."""
+    out = {}
+    for r in conn.execute(
+        """
+        SELECT fi.source_row_id AS srid, o.stage AS stage
+        FROM opportunities o
+        JOIN fundraising_investors fi ON o.fundraising_investor_id = fi.id
+        WHERE o.deleted_at IS NULL
+        """
+    ).fetchall():
+        srid = str(r['srid'] or '')
+        if srid:
+            out[srid] = r['stage']
+    return out
+
 def maybe_run_scheduled_backup():
     conn = get_db()
     try:
@@ -2066,6 +2144,10 @@ class CRMHandler(BaseHTTPRequestHandler):
             return self.handle_create_feature_request(user, body)
         if path == '/api/fundraising/log-communication':
             return self.handle_log_fundraising_communication(user, body)
+        if path == '/api/fundraising/pipeline/link':
+            return self.handle_pipeline_link(user, body)
+        if path == '/api/fundraising/pipeline/unlink':
+            return self.handle_pipeline_unlink(user, body)
         if path == '/api/fundraising/collab/heartbeat':
             return self.handle_fundraising_collab_heartbeat(user, body)
         if path == '/api/admin/users':
@@ -3066,6 +3148,152 @@ class CRMHandler(BaseHTTPRequestHandler):
         conn.close()
         return self.send_json({"data": {"communication": comm, "row": target_row, "version": next_version}}, 201)
 
+    def _fetch_opportunity_row(self, conn, opp_id):
+        return row_to_dict(conn.execute("""
+            SELECT op.*, c.first_name, c.last_name, c.email as contact_email,
+                   o.name as organization_name, u.full_name as owner_name
+            FROM opportunities op
+            LEFT JOIN contacts c ON op.contact_id = c.id
+            LEFT JOIN organizations o ON op.organization_id = o.id
+            LEFT JOIN users u ON op.owner_id = u.id
+            WHERE op.id = ?
+        """, (opp_id,)).fetchone())
+
+    def _resolve_grid_primary_contact(self, conn, investor_id, contact_index, actor_user_id):
+        """Resolve the classic contacts.id for a grid investor's chosen contact pill,
+        reusing the link sync already records in fundraising_contacts.contact_id (no
+        bare POST /api/contacts side-door). contact_index matches the pill order
+        (fundraising_contacts.sort_order)."""
+        try:
+            idx = int(contact_index)
+        except (TypeError, ValueError):
+            idx = 0
+        if idx < 0:
+            idx = 0
+        fc = conn.execute(
+            "SELECT contact_id, full_name, email FROM fundraising_contacts "
+            "WHERE investor_id = ? ORDER BY sort_order, rowid LIMIT 1 OFFSET ?",
+            (investor_id, idx)
+        ).fetchone()
+        if not fc:
+            # requested index out of range — fall back to the first pill
+            fc = conn.execute(
+                "SELECT contact_id, full_name, email FROM fundraising_contacts "
+                "WHERE investor_id = ? ORDER BY sort_order, rowid LIMIT 1",
+                (investor_id,)
+            ).fetchone()
+        if not fc:
+            return None
+        if fc['contact_id']:
+            return fc['contact_id']
+        # Rows predating the contact_id backfill: resolve via the same grid→classic
+        # upsert the relational sync uses, not a fresh side-door create.
+        inv = conn.execute("SELECT investor_name FROM fundraising_investors WHERE id = ?", (investor_id,)).fetchone()
+        investor_name = str(inv['investor_name'] if inv else '') or ''
+        return _upsert_contact_from_fundraising(
+            conn, investor_name, {"name": fc['full_name'], "email": fc['email']}, actor_user_id=actor_user_id
+        )
+
+    def handle_pipeline_link(self, user, body):
+        """Create (or return the existing) Pipeline opportunity for a fundraising-grid
+        investor row and link it durably via opportunities.fundraising_investor_id.
+        Idempotent: one live opp per investor — a re-link returns the existing opp
+        without reseeding its Pipeline-owned funnel fields."""
+        source_row_id = str(body.get('source_row_id') or '').strip()
+        if not source_row_id:
+            return self.send_error_json("source_row_id is required")
+
+        conn = get_db()
+        investor = conn.execute(
+            "SELECT id, investor_name, lead FROM fundraising_investors WHERE source_row_id = ?",
+            (source_row_id,)
+        ).fetchone()
+        if not investor:
+            conn.close()
+            return self.send_error_json("Investor not found for that grid row — save the grid first", 404)
+        investor_id = investor['id']
+        investor_name = str(investor['investor_name'] or '').strip() or 'Untitled Investor'
+
+        existing = conn.execute(
+            "SELECT id FROM opportunities WHERE fundraising_investor_id = ? AND deleted_at IS NULL "
+            "ORDER BY created_at LIMIT 1",
+            (investor_id,)
+        ).fetchone()
+        if existing:
+            opp = self._fetch_opportunity_row(conn, existing['id'])
+            conn.close()
+            return self.send_json({"data": opp, "already_linked": True})
+
+        contact_id = self._resolve_grid_primary_contact(
+            conn, investor_id, body.get('contact_index'), actor_user_id=user['user_id']
+        )
+        if not contact_id:
+            conn.close()
+            return self.send_error_json("Add at least one contact to the investor row before adding it to the pipeline")
+
+        contact = conn.execute("SELECT organization_id FROM contacts WHERE id = ?", (contact_id,)).fetchone()
+        org_id = contact['organization_id'] if contact else None
+
+        stage = str(body.get('stage') or 'lead').strip() or 'lead'
+        if stage not in PIPELINE_STAGES:
+            conn.close()
+            return self.send_error_json(f"Invalid stage. Must be one of: {', '.join(PIPELINE_STAGES)}")
+        try:
+            expected_amount = float(body.get('expected_amount') or 0)
+        except (TypeError, ValueError):
+            expected_amount = 0.0
+        try:
+            probability = int(body.get('probability'))
+        except (TypeError, ValueError):
+            probability = 35
+        probability = max(0, min(100, probability))
+        fund_name = str(body.get('fund_name') or '').strip()
+        name = str(body.get('name') or '').strip() or f"{investor_name} — Pipeline"
+        owner_id = _resolve_owner_from_lead(conn, investor['lead'], user['user_id'])
+
+        opp_id = generate_id()
+        conn.execute("""
+            INSERT INTO opportunities (id, name, contact_id, organization_id, stage,
+                commitment_amount, expected_amount, probability, fund_name,
+                owner_id, priority, fundraising_investor_id)
+            VALUES (?, ?, ?, ?, ?, 0, ?, ?, ?, ?, 'medium', ?)
+        """, (opp_id, name, contact_id, org_id, stage, expected_amount, probability,
+              fund_name or None, owner_id, investor_id))
+        log_audit(conn, user['user_id'], 'opportunity', opp_id, 'create', {"source": "grid_pipeline_link"})
+        conn.commit()
+
+        opp = self._fetch_opportunity_row(conn, opp_id)
+        conn.close()
+        return self.send_json({"data": opp, "already_linked": False}, 201)
+
+    def handle_pipeline_unlink(self, user, body):
+        """Remove a grid investor from the Pipeline: soft-delete its linked opportunity.
+        The fundraising-grid row (investor, contacts, commitments, notes) is left fully
+        intact — only the opportunity is archived (recoverable via the board)."""
+        source_row_id = str(body.get('source_row_id') or '').strip()
+        if not source_row_id:
+            return self.send_error_json("source_row_id is required")
+        conn = get_db()
+        investor = conn.execute(
+            "SELECT id FROM fundraising_investors WHERE source_row_id = ?", (source_row_id,)
+        ).fetchone()
+        if not investor:
+            conn.close()
+            return self.send_error_json("Investor not found for that grid row", 404)
+        opp_rows = conn.execute(
+            "SELECT id FROM opportunities WHERE fundraising_investor_id = ? AND deleted_at IS NULL",
+            (investor['id'],)
+        ).fetchall()
+        archived = 0
+        for opp in opp_rows:
+            conn.execute("UPDATE opportunities SET deleted_at = ?, updated_at = ? WHERE id = ?",
+                         (now(), now(), opp['id']))
+            log_audit(conn, user['user_id'], 'opportunity', opp['id'], 'delete', {"source": "grid_pipeline_unlink"})
+            archived += 1
+        conn.commit()
+        conn.close()
+        return self.send_json({"data": {"archived": archived}})
+
     def handle_intake_match(self, user, params):
         """Read-only: does an investor matching this intake already exist? Used by the
         Matrix intake bot to label its in-thread proposal new-vs-existing. Returns the
@@ -3154,11 +3382,11 @@ class CRMHandler(BaseHTTPRequestHandler):
         ).fetchone()['total']
 
         pipeline_value = conn.execute(
-            "SELECT COALESCE(SUM(expected_amount), 0) as total FROM opportunities WHERE stage NOT IN ('funded', 'lost')"
+            "SELECT COALESCE(SUM(expected_amount), 0) as total FROM opportunities WHERE stage NOT IN ('funded', 'lost') AND deleted_at IS NULL"
         ).fetchone()['total']
 
         active_opportunities = conn.execute(
-            "SELECT COUNT(*) as c FROM opportunities WHERE stage NOT IN ('funded', 'lost')"
+            "SELECT COUNT(*) as c FROM opportunities WHERE stage NOT IN ('funded', 'lost') AND deleted_at IS NULL"
         ).fetchone()['c']
 
         # Pipeline by stage
@@ -3166,7 +3394,7 @@ class CRMHandler(BaseHTTPRequestHandler):
             SELECT stage, COUNT(*) as count, COALESCE(SUM(expected_amount), 0) as total_value,
                    COALESCE(SUM(commitment_amount), 0) as committed_value
             FROM opportunities
-            WHERE stage != 'lost'
+            WHERE stage != 'lost' AND deleted_at IS NULL
             GROUP BY stage
             ORDER BY CASE stage
                 WHEN 'lead' THEN 1 WHEN 'outreach' THEN 2 WHEN 'meeting' THEN 3
@@ -3243,6 +3471,7 @@ class CRMHandler(BaseHTTPRequestHandler):
                    COALESCE(SUM(commitment_amount), 0) as total_committed,
                    COALESCE(AVG(probability), 0) as avg_probability
             FROM opportunities
+            WHERE deleted_at IS NULL
             GROUP BY stage
             ORDER BY CASE stage
                 WHEN 'lead' THEN 1 WHEN 'outreach' THEN 2 WHEN 'meeting' THEN 3
@@ -3255,6 +3484,7 @@ class CRMHandler(BaseHTTPRequestHandler):
                    COALESCE(SUM(op.expected_amount), 0) as total_expected
             FROM opportunities op
             LEFT JOIN users u ON op.owner_id = u.id
+            WHERE op.deleted_at IS NULL
             GROUP BY op.owner_id, op.stage
             ORDER BY u.full_name, op.stage
         """).fetchall())
@@ -3263,7 +3493,7 @@ class CRMHandler(BaseHTTPRequestHandler):
             SELECT priority, COUNT(*) as count,
                    COALESCE(SUM(expected_amount), 0) as total_expected
             FROM opportunities
-            WHERE stage NOT IN ('funded', 'lost')
+            WHERE stage NOT IN ('funded', 'lost') AND deleted_at IS NULL
             GROUP BY priority
         """).fetchall())
 
@@ -4824,6 +5054,7 @@ class CRMHandler(BaseHTTPRequestHandler):
         conn = get_db()
         self._ensure_fundraising_state_row(conn)
         row = conn.execute("SELECT * FROM fundraising_state WHERE id = 'main'").fetchone()
+        stage_by_row = pipeline_stage_by_source_row(conn)
         conn.close()
 
         try:
@@ -4842,6 +5073,16 @@ class CRMHandler(BaseHTTPRequestHandler):
         columns = grid.get('columns', [])
         rows = grid.get('rows', [])
 
+        # Inject the read-only pipeline columns, derived from the live linked opportunity
+        # (the opportunities table is canonical — never stored in the grid blob, so it
+        # can't go stale). The frontend renders these read-only and strips them on save.
+        for r in rows:
+            if not isinstance(r, dict):
+                continue
+            stage = stage_by_row.get(str(r.get('id') or ''))
+            r['pipeline'] = bool(stage)
+            r['pipeline_stage'] = stage or ''
+
         return self.send_json({
             "data": {
                 "grid": {"columns": columns, "rows": rows},
@@ -5020,6 +5261,9 @@ class CRMHandler(BaseHTTPRequestHandler):
             WHERE id = 'main'
         """, (json.dumps(grid), json.dumps(next_views), next_version, user['user_id'], now()))
         sync_fundraising_relational(conn, grid, next_views, actor_user_id=user['user_id'])
+        # Archive pipeline opps orphaned by an investor deleted from the grid (one-way
+        # cleanup; never creates or reseeds — see reconcile_grid_pipeline_links).
+        reconcile_grid_pipeline_links(conn)
         log_audit(conn, user['user_id'], 'fundraising_state', 'main', 'update', {"version": next_version})
         conn.commit()
         conn.close()

backend/test_grid_pipeline_link.py

@@ -0,0 +1,270 @@
+#!/usr/bin/env python3
+"""Tests for the grid → Pipeline link ("Adopt the Pipeline", v0.1.0:87).
+
+Boots the REAL server against a temp DB and exercises the new endpoints end-to-end:
+  - POST /api/fundraising/pipeline/link creates exactly ONE opportunity, linked via
+    opportunities.fundraising_investor_id, reusing the grid's synced contact (no
+    POST /api/contacts side-door) and mapping the grid 'lead' -> owner;
+  - the link is idempotent: a re-link returns the existing opp and NEVER reseeds its
+    Pipeline-owned funnel fields (stage/probability) — the board owns those;
+  - GET /api/fundraising/state injects read-only pipeline / pipeline_stage row values
+    derived from the live opp;
+  - linking a contactless row, or an unknown row, is refused;
+  - POST .../unlink soft-deletes the opp (off the board, recoverable) while leaving the
+    grid investor row fully intact;
+  - deleting an investor from the grid archives its orphaned opp on the next save;
+  - the pipeline report + dashboard aggregates exclude archived (soft-deleted) opps.
+Synthetic data only.
+
+Run: cd backend && python3 test_grid_pipeline_link.py
+"""
+import http.client
+import json
+import os
+import sqlite3
+import sys
+import tempfile
+import threading
+from http.server import ThreadingHTTPServer
+
+_DATA = tempfile.mkdtemp()
+os.environ["CRM_DATA_DIR"] = _DATA
+os.environ["CRM_DB_PATH"] = os.path.join(_DATA, "crm.db")
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+import server  # noqa: E402
+
+FAILS = []
+
+
+def check(cond, msg):
+    print(("  PASS " if cond else "  FAIL ") + msg)
+    if not cond:
+        FAILS.append(msg)
+
+
+class _Quiet(server.CRMHandler):
+    def log_message(self, *a):
+        pass
+
+
+def _req(port, method, path, token=None, body=None):
+    conn = http.client.HTTPConnection("127.0.0.1", port, timeout=10)
+    headers = {}
+    if token:
+        headers["Authorization"] = "Bearer " + token
+    payload = None
+    if body is not None:
+        payload = json.dumps(body)
+        headers["Content-Type"] = "application/json"
+    conn.request(method, path, body=payload, headers=headers)
+    resp = conn.getresponse()
+    raw = resp.read().decode("utf-8", "replace")
+    conn.close()
+    data = None
+    if raw:
+        try:
+            data = json.loads(raw)
+        except ValueError:
+            pass
+    return resp.status, data
+
+
+def _put_grid(port, token, rows):
+    return _req(port, "PUT", "/api/fundraising/state", token,
+                {"grid": {"columns": [], "rows": rows}, "views": []})
+
+
+ROW_ACME = {"id": "rowAcme", "investor_name": "Acme Capital", "notes": "", "lead": "Grant",
+            "contacts": [{"name": "Jane Doe", "email": "jane@acme.com", "title": "GP"}]}
+ROW_BETA = {"id": "rowBeta", "investor_name": "Beta Capital LLC", "notes": "", "lead": "",
+            "contacts": [{"name": "Pat Roe", "email": "pat@beta.com", "title": ""}]}
+ROW_EMPTY = {"id": "rowEmpty", "investor_name": "Empty LP", "notes": "", "contacts": []}
+
+
+def _db():
+    return sqlite3.connect(os.environ["CRM_DB_PATH"])
+
+
+def seed():
+    c = _db()
+    c.execute("INSERT INTO users (id,username,email,password_hash,full_name,role,is_active) "
+              "VALUES ('u1','grant','grant@ten31.example','x','Grant','admin',1)")
+    c.commit()
+    c.close()
+
+
+def _opp_count_live(fr_investor_id=None):
+    c = _db()
+    if fr_investor_id:
+        n = c.execute("SELECT COUNT(*) FROM opportunities WHERE fundraising_investor_id = ? "
+                      "AND deleted_at IS NULL", (fr_investor_id,)).fetchone()[0]
+    else:
+        n = c.execute("SELECT COUNT(*) FROM opportunities WHERE deleted_at IS NULL").fetchone()[0]
+    c.close()
+    return n
+
+
+def main():
+    server.init_db()
+    seed()
+    token = server.create_token("u1", "grant", "admin")
+
+    httpd = ThreadingHTTPServer(("127.0.0.1", 0), _Quiet)
+    port = httpd.server_address[1]
+    threading.Thread(target=httpd.serve_forever, daemon=True).start()
+    try:
+        st, _ = _put_grid(port, token, [ROW_ACME, ROW_BETA, ROW_EMPTY])
+        check(st == 200, f"seed grid via PUT /state (got {st})")
+
+        # ── link creates one linked opp with the seeds + resolved contact + mapped owner ──
+        print("\n[link: creates one linked opportunity with seeds]")
+        st, d = _req(port, "POST", "/api/fundraising/pipeline/link", token, {
+            "source_row_id": "rowAcme", "fund_name": "Fund III",
+            "expected_amount": 250000, "probability": 40, "stage": "outreach",
+        })
+        opp = (d or {}).get("data") or {}
+        check(st == 201 and (d or {}).get("already_linked") is False, f"link -> 201 new (got {st}, {d})")
+        check(opp.get("stage") == "outreach" and opp.get("expected_amount") == 250000
+              and opp.get("probability") == 40 and opp.get("fund_name") == "Fund III",
+              f"seeds applied (got {{stage:{opp.get('stage')}, amt:{opp.get('expected_amount')}, "
+              f"prob:{opp.get('probability')}, fund:{opp.get('fund_name')}}})")
+        check(opp.get("first_name") == "Jane", f"reused synced contact Jane Doe (got {opp.get('first_name')})")
+        check(opp.get("owner_name") == "Grant", f"grid lead 'Grant' -> owner Grant (got {opp.get('owner_name')})")
+        fr_id = opp.get("fundraising_investor_id")
+        check(bool(fr_id), f"opportunity carries fundraising_investor_id (got {fr_id})")
+        check(_opp_count_live(fr_id) == 1, "exactly one live opp linked to the investor")
+        opp_id = opp.get("id")
+
+        # ── idempotent re-link: returns existing, board-owned stage NOT reseeded ──
+        print("\n[idempotent: re-link returns existing opp without reseeding funnel fields]")
+        st, _ = _req(port, "PATCH", f"/api/opportunities/{opp_id}/stage", token, {"stage": "meeting"})
+        check(st == 200, f"advance stage on the board -> meeting (got {st})")
+        st, d = _req(port, "POST", "/api/fundraising/pipeline/link", token, {
+            "source_row_id": "rowAcme", "stage": "lead", "expected_amount": 999, "probability": 5,
+        })
+        opp2 = (d or {}).get("data") or {}
+        check(st == 200 and (d or {}).get("already_linked") is True, f"re-link -> already_linked (got {st}, {d})")
+        check(opp2.get("stage") == "meeting" and opp2.get("expected_amount") == 250000,
+              f"funnel fields preserved, not reseeded (got stage={opp2.get('stage')}, amt={opp2.get('expected_amount')})")
+        check(_opp_count_live(fr_id) == 1, "still exactly one live opp (no duplicate)")
+
+        # ── read-injection: GET state shows pipeline flag + stage, derived live ──
+        print("\n[read-injection: GET /state exposes read-only pipeline + pipeline_stage]")
+        st, d = _req(port, "GET", "/api/fundraising/state", token)
+        rows = {r["id"]: r for r in (d or {}).get("data", {}).get("grid", {}).get("rows", [])}
+        check(rows.get("rowAcme", {}).get("pipeline") is True
+              and rows.get("rowAcme", {}).get("pipeline_stage") == "meeting",
+              f"rowAcme pipeline true @meeting (got {rows.get('rowAcme', {}).get('pipeline')}, "
+              f"{rows.get('rowAcme', {}).get('pipeline_stage')})")
+        check(rows.get("rowBeta", {}).get("pipeline") is False
+              and rows.get("rowBeta", {}).get("pipeline_stage") == "",
+              f"rowBeta not in pipeline (got {rows.get('rowBeta', {}).get('pipeline')})")
+
+        # ── round-trip: a save echoing the injected read-only values is lossless ──
+        print("\n[round-trip: PUT carrying injected pipeline values strips them, link intact]")
+        st, d = _req(port, "GET", "/api/fundraising/state", token)
+        echoed = (d or {}).get("data", {}).get("grid", {}).get("rows", [])
+        st, _ = _put_grid(port, token, echoed)  # as the frontend autosave would, rows still carry pipeline*
+        check(st == 200, f"echo-back save -> 200 (got {st})")
+        check(_opp_count_live(fr_id) == 1, "link survives the round-trip (no dup, not archived)")
+        c = _db()
+        blob = json.loads(c.execute("SELECT grid_json FROM fundraising_state WHERE id='main'").fetchone()[0])
+        c.close()
+        stored_acme = {r["id"]: r for r in blob.get("rows", [])}.get("rowAcme", {})
+        check("pipeline" not in stored_acme and "pipeline_stage" not in stored_acme,
+              "computed keys are NOT persisted into the grid blob")
+        st, d = _req(port, "GET", "/api/fundraising/state", token)
+        rt = {r["id"]: r for r in (d or {}).get("data", {}).get("grid", {}).get("rows", [])}.get("rowAcme", {})
+        check(rt.get("pipeline") is True and rt.get("pipeline_stage") == "meeting",
+              f"pipeline values re-injected after round-trip (got {rt.get('pipeline')}, {rt.get('pipeline_stage')})")
+
+        # ── guards ──
+        print("\n[guard: a contactless row cannot be added to the pipeline]")
+        st, d = _req(port, "POST", "/api/fundraising/pipeline/link", token, {"source_row_id": "rowEmpty"})
+        check(st == 400, f"no contact -> 400 (got {st}, {d})")
+        check(_opp_count_live() == 1, "no stray opp created for the contactless row")
+
+        print("\n[guard: unknown grid row -> 404]")
+        st, _ = _req(port, "POST", "/api/fundraising/pipeline/link", token, {"source_row_id": "nope"})
+        check(st == 404, f"unknown row -> 404 (got {st})")
+
+        print("\n[guard: unauthenticated -> 401]")
+        st, _ = _req(port, "POST", "/api/fundraising/pipeline/link", None, {"source_row_id": "rowAcme"})
+        check(st == 401, f"no token -> 401 (got {st})")
+
+        # ── the opp loads on the board + counts in the dashboard while live ──
+        print("\n[board + dashboard count the live opp]")
+        st, d = _req(port, "GET", "/api/opportunities?limit=1000", token)
+        ids = [o["id"] for o in (d or {}).get("data", [])]
+        check(opp_id in ids, "linked opp appears on the board")
+        st, d = _req(port, "GET", "/api/reports/dashboard", token)
+        active = (d or {}).get("data", {}).get("metrics", {}).get("active_opportunities")
+        check(active == 1, f"dashboard active_opportunities == 1 (got {active})")
+
+        # ── unlink soft-deletes the opp; the GRID ROW stays fully intact ──
+        print("\n[unlink: archives the opp, leaves the grid investor intact]")
+        st, d = _req(port, "POST", "/api/fundraising/pipeline/unlink", token, {"source_row_id": "rowAcme"})
+        check(st == 200 and (d or {}).get("data", {}).get("archived") == 1, f"unlink -> archived 1 (got {st}, {d})")
+        check(_opp_count_live(fr_id) == 0, "opp is no longer live (soft-deleted)")
+        c = _db()
+        gone = c.execute("SELECT deleted_at FROM opportunities WHERE id = ?", (opp_id,)).fetchone()[0]
+        inv_still = c.execute("SELECT investor_name FROM fundraising_investors WHERE source_row_id = 'rowAcme'").fetchone()
+        contact_still = c.execute("SELECT COUNT(*) FROM fundraising_contacts WHERE investor_id = ?", (fr_id,)).fetchone()[0]
+        c.close()
+        check(gone is not None, "opp row tombstoned (deleted_at set), not hard-deleted")
+        check(inv_still and inv_still[0] == "Acme Capital", "grid investor row untouched by unlink")
+        check(contact_still >= 1, "grid investor's contacts untouched by unlink")
+        st, d = _req(port, "GET", "/api/opportunities?limit=1000", token)
+        check(opp_id not in [o["id"] for o in (d or {}).get("data", [])], "archived opp left the board")
+        st, d = _req(port, "GET", "/api/fundraising/state", token)
+        rows = {r["id"]: r for r in (d or {}).get("data", {}).get("grid", {}).get("rows", [])}
+        check(rows.get("rowAcme", {}).get("pipeline") is False, "grid no longer flags rowAcme as in-pipeline")
+
+        # ── aggregates exclude the archived opp ──
+        print("\n[aggregates exclude archived opps]")
+        st, d = _req(port, "GET", "/api/reports/dashboard", token)
+        active = (d or {}).get("data", {}).get("metrics", {}).get("active_opportunities")
+        check(active == 0, f"dashboard active_opportunities back to 0 (got {active})")
+        st, d = _req(port, "GET", "/api/reports/pipeline", token)
+        by_stage = (d or {}).get("data", {}).get("by_stage", [])
+        total = sum(s.get("count", 0) for s in by_stage)
+        check(total == 0, f"pipeline report by_stage excludes archived (got total {total})")
+
+        # ── re-link after unlink: a fresh opp is created (the archived one stays archived) ──
+        print("\n[re-link after unlink: creates a new opp, flag reappears]")
+        st, d = _req(port, "POST", "/api/fundraising/pipeline/link", token, {
+            "source_row_id": "rowAcme", "stage": "outreach", "expected_amount": 50000,
+        })
+        relinked = (d or {}).get("data") or {}
+        check(st == 201 and (d or {}).get("already_linked") is False and relinked.get("id") != opp_id,
+              f"re-link -> a NEW opp distinct from the archived one (got {st}, {relinked.get('id')} vs {opp_id})")
+        check(_opp_count_live(fr_id) == 1, "exactly one live opp again after re-link")
+        st, _ = _req(port, "POST", "/api/fundraising/pipeline/unlink", token, {"source_row_id": "rowAcme"})
+        check(st == 200, "reset: unlink the re-linked opp")
+
+        # ── orphan reconciler: deleting the investor from the grid archives its opp ──
+        print("\n[orphan: deleting the grid investor archives its linked opp on next save]")
+        st, d = _req(port, "POST", "/api/fundraising/pipeline/link", token, {
+            "source_row_id": "rowBeta", "stage": "lead", "expected_amount": 100000,
+        })
+        beta = (d or {}).get("data") or {}
+        beta_opp_id, beta_fr = beta.get("id"), beta.get("fundraising_investor_id")
+        check(st == 201 and _opp_count_live(beta_fr) == 1, f"beta linked (got {st})")
+        # drop rowBeta from the grid (keep the others)
+        st, _ = _put_grid(port, token, [ROW_ACME, ROW_EMPTY])
+        check(st == 200, f"save grid without rowBeta (got {st})")
+        check(_opp_count_live(beta_fr) == 0, "beta's orphaned opp archived by the reconciler")
+        st, d = _req(port, "GET", "/api/opportunities?limit=1000", token)
+        check(beta_opp_id not in [o["id"] for o in (d or {}).get("data", [])], "orphaned opp left the board")
+    finally:
+        httpd.shutdown()
+
+    print("\n" + ("ALL PASS" if not FAILS else f"{len(FAILS)} FAILURE(S):"))
+    for f in FAILS:
+        print("  - " + f)
+    sys.exit(1 if FAILS else 0)
+
+
+if __name__ == "__main__":
+    main()