feat(notion): Ops Copilot ← Notion Command Center bridge + gated lead sync

config.yaml

@@ -176,3 +176,17 @@ scraper:
     price: ".product-price"
     image: "img"
     link: "a"
+
+# Notion "Command Center" integration. DB ids are identifiers, NOT secrets — the
+# only secret is NOTION_TOKEN (Secret Manager). The Ops Copilot reads these
+# PII-free (count-by-status) when NOTION_COMMAND_CENTER=on. Each id can also be
+# overridden by env NOTION_<KEY>_DB_ID if you'd rather not commit them.
+notion:
+  databases:
+    delivery_tracker: "34e6688d-bec6-81b7-966b-c7bfebeb7a27"   # 🔧 Delivery Tracker (installations)
+    cs_survey: "34e6688d-bec6-818b-abd5-f6a095be6cb9"          # 🤝 Customer satisfaction surveys
+    service_warranty: "34e6688d-bec6-8190-9a2c-dc0f5db19de6"   # 🛠️ Service & Warranty
+    title: "34e6688d-bec6-81db-b2a6-ddf06dabe151"              # 📋 Title processing
+    collections: "34e6688d-bec6-81c8-8430-ec5b1d2efd6d"        # 💰 Collections
+    insurance: "34e6688d-bec6-8196-84a6-fc3c64ac2768"          # Insurance / KIP
+    lead_pipeline: "34e6688d-bec6-81b3-b3a3-d176c0f7ce44"      # 👤 Lead Pipeline ("up log")

docs/INTEGRATION_GCP_ACTIVATION.md

@@ -0,0 +1,100 @@
+# THO Integration — GCP Activation Runbook
+
+Staged, audited commands to take the inventory + Notion-bridge integration live.
+Every step here is a **gated production action** (it deploys/configures the live
+client app), so it is yours to run — the code is built, tested, and reversible;
+this just makes the activation push-button.
+
+> **Audited prod state (2026-06-16):** service `project-go-forward` @ rev `00279`,
+> project `tho-ai-agent`, region `us-central1`, run-as SA
+> `691674245427-compute@developer.gserviceaccount.com`, Firestore `(default)`
+> NATIVE us-central1. **All integration env vars are currently unset**
+> (`INVENTORY_SOURCE`, `NOTION_TOKEN`, `NOTION_COMMAND_CENTER`, …) → the feature
+> is fully dark; nothing changes until you run the steps below.
+
+> ⚠️ **Golden rule for Cloud Run config:** always use `--update-env-vars` /
+> `--update-secrets` (additive). **Never** `--set-env-vars` / `--set-secrets` —
+> that would wipe the 16 existing env vars + 7 secret mounts (WEBAUTHN_*, RESEND,
+> admin PIN/session, partner keys) and break prod.
+
+---
+
+## Part A — Inventory unfreeze  (branch `feat/in-app-inventory`)
+
+1. **Rebase + merge.** That branch was cut from main `6098a91` (#192); main is now
+   ≥ #198. Rebase onto current main (changes are additive — new files + targeted
+   edits, should be clean), then merge the PR. Default `INVENTORY_SOURCE=legacy`
+   → **the merge deploys with zero behavior change.**
+
+2. **Seed Firestore from the real source.** Put `House Orders.xlsx` in `data/`,
+   preview, then write (uses Application Default Credentials with Firestore access
+   to `tho-ai-agent`):
+   ```bash
+   python -m tools.house_orders_sync                 # dry-run: eyeball the homes + AVAILABLE/SOLD split
+   python -m tools.house_orders_sync --apply         # upsert into Firestore `inventory` (idempotent)
+   ```
+   (Fallback if you don't have the sheet handy: `python -m tools.inventory_seed --apply`
+   seeds the 279-home May-11 snapshot instead.)
+
+3. **Flip the public read onto the in-app store** (gated deploy → new revision):
+   ```bash
+   gcloud run services update project-go-forward --region=us-central1 --project=tho-ai-agent \
+     --update-env-vars INVENTORY_SOURCE=firestore
+   ```
+   The site now serves Mark's current stock. **Instant revert** (no redeploy delay,
+   just a new revision):
+   ```bash
+   gcloud run services update project-go-forward --region=us-central1 --project=tho-ai-agent \
+     --update-env-vars INVENTORY_SOURCE=legacy
+   ```
+
+---
+
+## Part B — Notion bridge / Ops Copilot  (branch `feat/notion-ops-bridge`)
+
+The 7 ops-DB ids are already in `config.yaml`; the only secret is the token.
+
+1. **Token secret — ✅ DONE.** Stored as Secret Manager secret **`notion-id`**
+   (version 1 enabled). To rotate later:
+   `printf '%s' 'ntn_NEW' | gcloud secrets versions add notion-id --data-file=- --project=tho-ai-agent`.
+
+2. **SA read access — ✅ DONE.** The run-as SA
+   `691674245427-compute@developer.gserviceaccount.com` was granted
+   `roles/secretmanager.secretAccessor` on `notion-id` (so the deploy below can bind it).
+
+3. **Share the 7 databases with the integration** (Notion UI, once each): on the
+   hub page (or each DB) → `•••` → **Connections** → add your integration. The
+   ids the app uses: Delivery Tracker, Customer-satisfaction surveys, Service &
+   Warranty, Title processing, Collections, Insurance/KIP, Lead Pipeline.
+
+4. **Merge `feat/notion-ops-bridge`**, then wire + flip on (gated deploy):
+   ```bash
+   gcloud run services update project-go-forward --region=us-central1 --project=tho-ai-agent \
+     --update-secrets NOTION_TOKEN=notion-id:latest \
+     --update-env-vars NOTION_COMMAND_CENTER=on
+   ```
+
+5. **Verify:** ask the Ops Copilot *"how many titles are pending?"* or *"how many
+   homes are in trim-out?"* — it answers from the staff's live Notion, **count-by-
+   status only, no customer PII**. **Instant revert:** `--update-env-vars NOTION_COMMAND_CENTER=off`
+   (the snapshot silently returns to Firestore counts).
+
+---
+
+## Part C — deferred GCP automation (built later, gated)
+
+- **Daily House Orders sync** — Cloud Scheduler → an admin sync endpoint →
+  `house_orders_sync` → Firestore, so the site refreshes without a manual run.
+  (Needs a sheet→GCS drop or Drive-API access for the run-as SA.)
+- **Mira-secret cleanup** — after the Ops Copilot bridge proves parity in prod,
+  retire `tho-api-key-mira` + `telegram-asfao-token` and delete the Mira routers
+  (`mira_routes.py`, `mira_notify.py`, `github_mira_trigger.py`).
+
+## Security notes
+- `NOTION_TOKEN` lives only in Secret Manager; DB ids are config (not secrets).
+- The bridge is **PII-free by construction** — count-by-status only; no customer
+  identity, dollar figure, or free text ever enters the snapshot.
+- Every activation is a single env var with an instant, no-code revert.
+- The two existing scheduler loops (`brain-ooda-loop`, `asfao-decide-loop`) were
+  audited: both write analysis/**proposed** decisions to BigQuery — neither
+  executes trades. The no-auto-execution boundary holds.

main.py

@@ -4694,6 +4694,30 @@ async def submit_contact_form(request: Request):
                 lead_id=lead_id,
             )
 
+        # Mirror the lead into the Notion Lead Pipeline the salespeople work from
+        # (server-side, gated by NOTION_LEAD_SYNC, fire-and-forget). PII goes only
+        # to the staff's own CRM that already holds contact info; this is never
+        # reachable from a public/partner route and never blocks the visitor.
+        try:
+            from tools.notion_lead_writer import is_lead_sync_enabled, write_lead
+
+            if is_lead_sync_enabled():
+                # write_lead is fire-and-forget and logs notion_lead_sync_failed
+                # itself on failure. Do NOT surface that to the anonymous visitor
+                # in `warnings` — keep the integration invisible externally.
+                write_lead(
+                    name,
+                    email=email or None,
+                    phone=phone,
+                    message=data.get("message"),
+                    source=data.get("source", "contact_form"),
+                    lead_id=lead_id,
+                )
+        except Exception as e:
+            struct_logger.warning(
+                "Notion lead sync failed", event="notion_lead_sync_failed", error=str(e)
+            )
+
         # Send welcome email if email provided
         if email:
             try:
@@ -6857,6 +6881,26 @@ def _seo_public_homes() -> list:
 from schemas.copilot_schemas import CopilotRequest
 
 
+@app.get("/api/admin/ops-snapshot", dependencies=[Depends(require_admin)])
+async def admin_ops_snapshot():
+    """Live, PII-free business snapshot for the admin Ops dashboard.
+
+    Read-only and admin-gated. Aggregates COUNTS only (leads/appointments/
+    inventory/deals/installations/feedback, plus the Notion Command Center ops
+    counts — delivery/title/collections/insurance — when NOTION_COMMAND_CENTER is
+    on). Never any customer identity or dollar figure. Each section is
+    fault-isolated; this never 500s. This is the read surface that replaces the
+    Telegram/Mira status pushes with an in-app, GCP-native view.
+    """
+    from tools.ops_copilot import get_business_snapshot
+
+    try:
+        return {"success": True, "snapshot": await get_business_snapshot()}
+    except Exception as e:
+        struct_logger.error("Ops snapshot failed", error=str(e))
+        return {"success": False, "error": "Snapshot unavailable."}
+
+
 @app.post("/api/admin/copilot", dependencies=[Depends(require_admin)])
 async def admin_ops_copilot(body: CopilotRequest):
     """Answer a staff question using live business data + platform how-to.

tests/test_api_v1.py

@@ -2459,3 +2459,24 @@ def test_create_inventory_strips_dealer_cost(monkeypatch):
     for forbidden in ("invoice_amount", "invoice_date", "cost"):
         assert forbidden not in stored
     assert stored["model_name"] == "The Nassau"
+
+
+def test_ops_snapshot_requires_admin(monkeypatch):
+    client, _main, _db, _logger = create_client(monkeypatch, tho_api_key="tho-secret")
+    assert client.get("/api/admin/ops-snapshot").status_code == 401
+
+
+def test_ops_snapshot_returns_counts(monkeypatch):
+    client, main, _db, _logger = create_client(monkeypatch, tho_api_key="tho-secret")
+    from tools import ops_copilot
+
+    async def _fake_snapshot():
+        return {"leads": {"total": 3}, "operations": {"title": {"by_status": {"Title Issued": 5}}}}
+
+    monkeypatch.setattr(ops_copilot, "get_business_snapshot", _fake_snapshot)
+    token = main._create_admin_token()
+    resp = client.get("/api/admin/ops-snapshot", headers={"X-Admin-Token": token})
+    assert resp.status_code == 200
+    body = resp.json()
+    assert body["success"] is True
+    assert body["snapshot"]["operations"]["title"]["by_status"] == {"Title Issued": 5}

tests/test_notion_client.py

@@ -242,3 +242,101 @@ def test_fetch_skips_non_dict_results(notion_env, monkeypatch):
     rows = nc.fetch_feedback()
     assert len(rows) == 1
     assert rows[0]["id"] == "page-fb-1"
+
+
+# --------------------------------------------------------------------------- #
+# Command Center reader (config-driven, GCP-native — increment 2)
+# --------------------------------------------------------------------------- #
+def _ops_page(status, *, prop_name="Status", with_pii=True):
+    props = {prop_name: _status(status)}
+    if with_pii:
+        # PII columns that must NEVER reach a count-by-status result:
+        props["Customer Name"] = _title("Jane Buyer")
+        props["Phone Number"] = {"type": "phone_number", "phone_number": "555-0000"}
+        props["Escrow Balance"] = _number(12345.67)
+    return {"id": f"pg-{status}", "created_time": "2026-06-14T10:00:00.000Z", "properties": props}
+
+
+def test_command_center_disabled_by_default(monkeypatch):
+    monkeypatch.delenv("NOTION_COMMAND_CENTER", raising=False)
+    assert nc.is_command_center_enabled() is False
+    for val in ("on", "true", "1", "YES"):
+        monkeypatch.setenv("NOTION_COMMAND_CENTER", val)
+        assert nc.is_command_center_enabled() is True
+    monkeypatch.setenv("NOTION_COMMAND_CENTER", "off")
+    assert nc.is_command_center_enabled() is False
+
+
+def test_db_id_resolves_config_then_env(monkeypatch):
+    # Configured keys resolve from config.yaml notion.databases.* (real ids).
+    assert nc._db_id("title").startswith("34e6688d")
+    assert nc._db_id("delivery_tracker").startswith("34e6688d")
+    # An unconfigured key (not in config.yaml) falls back to env.
+    monkeypatch.setenv("NOTION_TEAM_TASKS_DB_ID", "db-team")
+    assert nc._db_id("team_tasks") == "db-team"
+    assert nc._db_id("unknown_key") == ""
+
+
+def test_fetch_status_counts_counts_by_status_and_is_pii_free(monkeypatch):
+    monkeypatch.setenv("NOTION_TOKEN", "secret_test")
+    monkeypatch.setenv("NOTION_TITLE_DB_ID", "db-title")
+    _patch_post(monkeypatch, payload={"results": [
+        _ops_page("Title Issued"),
+        _ops_page("Title Issued"),
+        _ops_page("MCO Received from Factory"),
+    ]})
+    counts = nc.fetch_status_counts("title")
+    assert counts == {"Title Issued": 2, "MCO Received from Factory": 1}
+    # The PII columns on every page must not have leaked into the result.
+    blob = repr(counts).lower()
+    assert "jane" not in blob and "555-0000" not in blob and "12345" not in blob
+
+
+def test_fetch_status_counts_tolerates_dirty_and_aliased_status_columns(monkeypatch):
+    monkeypatch.setenv("NOTION_TOKEN", "secret_test")
+    monkeypatch.setenv("NOTION_DELIVERY_TRACKER_DB_ID", "db-delivery")
+    _patch_post(monkeypatch, payload={"results": [
+        _ops_page("Delivered to Site", prop_name="Current Phase"),     # alias, not "Status"
+        _ops_page("Delivered to Site", prop_name="Status "),            # trailing-space column
+        _ops_page("Unknown one", prop_name="No Status Column At All"),  # -> UNKNOWN
+    ]})
+    counts = nc.fetch_status_counts("delivery_tracker")
+    assert counts == {"Delivered to Site": 2, "UNKNOWN": 1}
+
+
+def test_fetch_status_counts_empty_when_unconfigured(monkeypatch):
+    # Use a key NOT in config.yaml so resolution depends only on env/token.
+    monkeypatch.delenv("NOTION_TOKEN", raising=False)
+    monkeypatch.setenv("NOTION_TEAM_TASKS_DB_ID", "db-team")
+    assert nc.fetch_status_counts("team_tasks") == {}  # no token
+    monkeypatch.setenv("NOTION_TOKEN", "secret_test")
+    monkeypatch.delenv("NOTION_TEAM_TASKS_DB_ID", raising=False)
+    assert nc.fetch_status_counts("team_tasks") == {}  # no db id (not in config or env)
+
+
+def test_query_database_paginates_across_has_more(monkeypatch):
+    """fetch_status_counts must count across pages, not silently cap at 100."""
+    monkeypatch.setenv("NOTION_TOKEN", "secret_test")
+    pages = [
+        {"results": [_ops_page("A"), _ops_page("A")], "has_more": True, "next_cursor": "cur2"},
+        {"results": [_ops_page("B"), _ops_page("A")], "has_more": False, "next_cursor": None},
+    ]
+    state = {"i": 0, "cursors": []}
+
+    def fake_post(url, headers=None, json=None, timeout=None):
+        state["cursors"].append(json.get("start_cursor"))
+        page = pages[state["i"]]
+        state["i"] += 1
+        return FakeResponse(page)
+
+    monkeypatch.setattr(nc.httpx, "post", fake_post)
+    counts = nc.fetch_status_counts("title")  # db id from config.yaml
+    assert counts == {"A": 3, "B": 1}          # summed across both pages
+    assert state["cursors"] == [None, "cur2"]   # page 2 used next_cursor
+
+
+def test_status_label_guard_collapses_pii_like_values():
+    assert nc._safe_status_label("Title Issued") == "Title Issued"          # enum-like: kept
+    assert nc._safe_status_label("Call John Smith 555-123-4567") == "OTHER"  # phone -> OTHER
+    assert nc._safe_status_label("email me at j@x.com") == "OTHER"           # email -> OTHER
+    assert nc._safe_status_label("x" * 80) == "OTHER"                        # overlong -> OTHER