feat: ABM Engine v1 — OpenSearch store + Apify enrichment + ICP scoring + k-NN memory + optional Neo4j mirror

- Aiven OpenSearch 3.6 (k-NN plugin): accounts/contacts/events/memory indices
- Apify: owner's own actors (google maps master scraper / linkedin company emails)
- ICP scoring: deterministic, explainable, 0-100 with breakdown (A/B/C/D labels)
- Vector memory: fastembed MiniLM-384 with hash-embedding fallback
- Graph mirror: BetterDB (managed Neo4j) or direct Aura, auto-detect, optional
- Cache: Upstash REST optional, in-process TTL fallback
- Auth: Supabase JWKS (same session JWT as validator) + optional API key
- Tests: 13 unit tests (normalizers, scoring, embeddings, auth, API contract)

Dockerfile

@@ -0,0 +1,18 @@
+FROM python:3.11-slim
+WORKDIR /app
+
+RUN apt-get update && apt-get install -y --no-install-recommends curl && rm -rf /var/lib/apt/lists/*
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+# Bake the embedding model at build time so a cold Space makes zero
+# network calls for inference (fastembed caches under HF_HOME).
+ENV HF_HOME=/app/models
+RUN python3 -c "from fastembed import TextEmbedding; list(TextEmbedding('sentence-transformers/all-MiniLM-L6-v2').embed(['warm']))" \
+    && echo "[BUILD] embedding model baked"
+
+COPY app/ ./app/
+
+EXPOSE 7860
+CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "7860"]

README.md

@@ -1,10 +1,37 @@
 ---
-title: Abm Engine
-emoji: 💻
+title: ABM Engine
+emoji: 🎯
 colorFrom: blue
-colorTo: gray
+colorTo: green
 sdk: docker
+app_port: 7860
 pinned: false
 ---
+# FortyDN ABM Engine
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+Outbound-intelligence service: account store + Apify enrichment + ICP scoring
++ vector memory + optional Neo4j graph mirror.
+
+**Stack**: FastAPI on Hugging Face Space · Aiven OpenSearch 3.6 (store, search,
+k-NN vectors) · Apify actors (owner's own scrapers) · BetterDB Neo4j (optional
+mirror) · Upstash Redis (optional cache) · Supabase JWKS identity.
+
+## Endpoints (under /api/v1)
+| Route | Purpose |
+|---|---|
+| GET /health | component report (open) |
+| POST /accounts | upsert account (domain-keyed) |
+| GET /accounts?q= | search with filters |
+| GET /accounts/{id} | account + contacts + events |
+| POST /enrich | run Apify actor → normalize → score → embed → mirror |
+| POST /signals/score | rescore an account |
+| POST /memory, /memory/batch | remember texts (384-dim embeddings) |
+| POST /memory/search | k-NN recall |
+| GET /graph/overview | Neo4j mirror status |
+
+## Secrets (Space settings)
+`OS_URL`, `OS_USER`, `OS_PASSWORD` (Aiven OpenSearch) · `APIFY_TOKEN` ·
+`SUPABASE_URL` · optional `BETTERDB_URL`+`BETTERDB_TOKEN` or
+`NEO4J_URI`+`NEO4J_USER`+`NEO4J_PASSWORD` · optional
+`UPSTASH_REDIS_REST_URL`+`UPSTASH_REDIS_REST_TOKEN` · optional `API_KEY`
+(service calls) · `LICENSE_ENFORCEMENT=true` at go-live.

app/auth.py

@@ -0,0 +1,119 @@
+"""Supabase JWT verification — same identity plane as the validator engine.
+
+The ABM engine is a product backend: the site attaches the user's Supabase
+session JWT (Authorization: Bearer) and this module verifies it against the
+project's public JWKS endpoint. Mirrors app/auth.py of modul-x (ES256/RS256,
+audience "authenticated"). A static X-API-Key (settings.api_key) is accepted
+for service-to-service calls; both are optional while enforcement is off.
+"""
+import logging
+from typing import Optional
+
+import httpx
+from fastapi import Request
+from fastapi.responses import JSONResponse
+from jose import jwt as jose_jwt
+from jose import jwk as jose_jwk
+from jose.exceptions import JWTError
+
+from .config import settings
+
+logger = logging.getLogger("abm.auth")
+
+_jwks_cache: Optional[dict] = None
+
+
+def enforcement_on() -> bool:
+    return (settings.license_enforcement or "").strip().lower() == "true"
+
+
+async def _get_jwks() -> dict:
+    global _jwks_cache
+    if _jwks_cache is None:
+        url = f"{settings.supabase_url}/auth/v1/.well-known/jwks.json"
+        async with httpx.AsyncClient(timeout=10.0) as client:
+            res = await client.get(url)
+            res.raise_for_status()
+            _jwks_cache = res.json()
+    return _jwks_cache  # type: ignore[return-value]
+
+
+async def verify_supabase_jwt(token: str) -> dict:
+    jwks = await _get_jwks()
+    last_error: Optional[Exception] = None
+    for key_data in jwks.get("keys", []):
+        try:
+            key = jose_jwk.construct(key_data)
+            claims = jose_jwt.decode(
+                token,
+                key,
+                algorithms=["ES256", "RS256"],
+                audience="authenticated",
+            )
+            if not claims.get("sub"):
+                raise JWTError("JWT has no sub claim")
+            return claims
+        except JWTError as exc:
+            last_error = exc
+            continue
+    raise JWTError(f"JWT verification failed: {last_error}")
+
+
+async def has_abm_license(user_id: str) -> bool:
+    """Active 'abm-engine' license in the central project (service-role read).
+    Falls back to modul-x license so existing customers are not locked out
+    during rollout."""
+    service_key = settings.supabase_service_role_key
+    if not service_key:
+        return False
+    headers = {"apikey": service_key, "Authorization": f"Bearer {service_key}"}
+    try:
+        async with httpx.AsyncClient(timeout=10.0) as client:
+            res = await client.get(
+                f"{settings.supabase_url}/rest/v1/licenses",
+                params={"user_id": f"eq.{user_id}", "status": "eq.active", "select": "product"},
+                headers=headers,
+            )
+        if res.status_code != 200:
+            return False
+        products = {row.get("product") for row in res.json()}
+        return bool(products & {"abm-engine", "modul-x", "combo-full-bridge"})
+    except Exception:
+        logger.warning("license lookup failed", exc_info=True)
+        return False
+
+
+def _api_key_ok(request: Request) -> bool:
+    supplied = request.headers.get("x-api-key", "")
+    return bool(settings.api_key) and supplied == settings.api_key
+
+
+async def authenticate(request: Request) -> dict:
+    """Authenticate a paid-route request.
+
+    Accepts, in order: X-API-Key (service calls), Supabase session JWT.
+    Returns {"user_id", "email", "via"}. Raises PermissionError on failure —
+    the route layer maps it to 401/402. When enforcement is off, an invalid
+    JWT still authenticates anonymously so demo deployments keep working.
+    """
+    if _api_key_ok(request):
+        return {"user_id": "service", "email": None, "via": "api-key"}
+
+    header = request.headers.get("authorization", "")
+    token = header[7:].strip() if header.lower().startswith("bearer ") else ""
+    if token and settings.supabase_url:
+        try:
+            claims = await verify_supabase_jwt(token)
+            user_id = claims.get("sub", "")
+            if enforcement_on() and not await has_abm_license(user_id):
+                raise PermissionError("402:No active ABM Engine license — see /pricing.")
+            return {"user_id": user_id, "email": claims.get("email"), "via": "jwt"}
+        except PermissionError:
+            raise
+        except JWTError:
+            if enforcement_on():
+                raise PermissionError("401:Invalid or expired token — sign in again.")
+            return {"user_id": "anonymous", "email": None, "via": "anonymous"}
+    if enforcement_on():
+        raise PermissionError("401:Missing bearer token — sign in first.")
+    return {"user_id": "anonymous", "email": None, "via": "anonymous"}

app/config.py

@@ -0,0 +1,82 @@
+# ABM Engine configuration — all optional at boot, degrade gracefully.
+import os
+from typing import List
+
+from pydantic import ConfigDict
+from pydantic_settings import BaseSettings
+
+
+class Settings(BaseSettings):
+    app_name: str = "FortyDN ABM Engine"
+    app_version: str = "1.0.0"
+    debug: bool = False
+
+    # ── Supabase identity (same project as the validator) ────────────────
+    supabase_url: str = ""
+    supabase_service_role_key: str = ""
+    license_enforcement: str = ""  # "true" gates paid routes on licenses
+
+    # ── Aiven OpenSearch (primary store + search + k-NN vectors) ─────────
+    os_url: str = ""
+    os_user: str = ""
+    os_password: str = ""
+    os_index_prefix: str = "abm"
+
+    # ── Apify enrichment ─────────────────────────────────────────────────
+    apify_token: str = ""
+    apify_actor_google_maps: str = "sitcod3.lab/fdn-google-map-master-scraper"
+    apify_actor_linkedin: str = "sitcod3.lab/linkedin-company-email-scraper"
+    apify_run_timeout_s: int = 900
+
+    # ── Optional graph mirror: BetterDB (managed Neo4j) or direct Aura ───
+    betterdb_url: str = ""
+    betterdb_token: str = ""
+    neo4j_uri: str = ""
+    neo4j_user: str = ""
+    neo4j_password: str = ""
+
+    # ── Optional Upstash Redis cache (REST) ─────────────────────────────
+    upstash_rest_url: str = ""
+    upstash_rest_token: str = ""
+
+    # ── Embeddings ───────────────────────────────────────────────────────
+    embedding_provider: str = "auto"  # auto | fastembed | hash
+    embedding_dim: int = 384
+
+    # ── ICP scoring defaults (override via env) ─────────────────────────
+    icp_industries: str = "software,saas,technology,manufacturing,logistics,healthcare,finance,professional services"
+    icp_countries: str = "united states,usa,canada,united kingdom,germany,netherlands,australia,denmark,sweden"
+    icp_min_headcount: int = 10
+    icp_max_headcount: int = 2000
+
+    # ── Access ───────────────────────────────────────────────────────────
+    api_key: str = ""  # optional X-API-Key for service-to-service calls
+
+    @property
+    def icp_industries_list(self) -> List[str]:
+        return [x.strip().lower() for x in self.icp_industries.split(",") if x.strip()]
+
+    @property
+    def icp_countries_list(self) -> List[str]:
+        return [x.strip().lower() for x in self.icp_countries.split(",") if x.strip()]
+
+    @property
+    def os_configured(self) -> bool:
+        return bool(self.os_url)
+
+    @property
+    def apify_configured(self) -> bool:
+        return bool(self.apify_token)
+
+    @property
+    def graph_configured(self) -> bool:
+        return bool((self.betterdb_url and self.betterdb_token) or self.neo4j_uri)
+
+    @property
+    def upstash_configured(self) -> bool:
+        return bool(self.upstash_rest_url and self.upstash_rest_token)
+
+    model_config = ConfigDict(env_file=".env", extra="ignore")
+
+
+settings = Settings()

app/main.py

@@ -0,0 +1,285 @@
+"""FortyDN ABM Engine — outbound intelligence service.
+
+Account graph + enrichment + ICP signals + vector memory, served as the
+product backend for the ABM module. Components (all optional at boot):
+
+  - Aiven OpenSearch : primary store, search, k-NN vector memory (required)
+  - Apify            : enrichment actors (Google Maps / LinkedIn)
+  - BetterDB / Aura  : optional Neo4j graph mirror
+  - Upstash Redis    : optional distributed cache
+  - Supabase JWKS    : user identity (same session JWT as the validator)
+
+Endpoints
+  GET  /health                 component report
+  POST /accounts               upsert account
+  GET  /accounts               search (?q=&industry=&country=&min_score=&limit=)
+  GET  /accounts/{id}          account + contacts + events
+  POST /enrich                 Apify enrichment run (google_maps | linkedin)
+  POST /signals/score          rescore an account
+  POST /memory                 remember a text document (embedded)
+  POST /memory/batch           remember many
+  POST /memory/search          k-NN recall
+  GET  /graph/overview         Neo4j mirror status/counts
+"""
+import logging
+from contextlib import asynccontextmanager
+from typing import Any, Dict, List, Optional
+
+from fastapi import FastAPI, HTTPException, Request
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import JSONResponse
+from pydantic import BaseModel, Field
+
+from .auth import authenticate
+from .config import settings
+from .services import accounts as accounts_svc
+from .services import enrich as enrich_svc
+from .services import memory as memory_svc
+from .services import signals as signals_svc
+from .stores import opensearch
+from .stores.apify import apify_client as apify
+from .stores.cache import cache
+from .stores.embeddings import provider_name
+from .stores.graph import graph
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger("abm.main")
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    logger.info("%s v%s starting", settings.app_name, settings.app_version)
+    if settings.os_configured:
+        try:
+            results = await opensearch.store.bootstrap()
+            logger.info("OpenSearch bootstrap: %s", results)
+            graph_ok = await graph.connect()
+            logger.info("graph mirror: %s", graph_ok)
+        except Exception as exc:
+            logger.warning("OpenSearch bootstrap failed (best-effort): %s", exc)
+    else:
+        logger.warning("OpenSearch not configured — set OS_URL/OS_USER/OS_PASSWORD")
+    yield
+    await opensearch.store.close()
+    await graph.close()
+    await apify.close()
+
+
+app = FastAPI(title=settings.app_name, version=settings.app_version, lifespan=lifespan)
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=False,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+
+@app.middleware("http")
+async def auth_gate(request: Request, call_next):
+    """Paid-route gate: /health is open; everything under /api requires
+    authentication. Enforcement off → anonymous demo access still works."""
+    path = request.url.path
+    if path == "/health" or path == "/" or not path.startswith("/api"):
+        return await call_next(request)
+    try:
+        request.state.auth = await authenticate(request)
+    except PermissionError as exc:
+        message = str(exc)
+        if message.startswith("402"):
+            return JSONResponse(status_code=402, content={"detail": message.split(":", 1)[1]})
+        return JSONResponse(status_code=401, content={"detail": message.split(":", 1)[-1]})
+    return await call_next(request)
+
+
+# All product routes live under /api/v1 (productApi.enginePath() convention).
+api_prefix = "/api/v1"
+
+
+class AccountIn(BaseModel):
+    domain: Optional[str] = None
+    name: Optional[str] = None
+    industry: Optional[str] = None
+    employee_count: Optional[int] = None
+    hq_country: Optional[str] = None
+    city: Optional[str] = None
+    website: Optional[str] = None
+    phone: Optional[str] = None
+    address: Optional[str] = None
+    description: Optional[str] = None
+    source: Optional[str] = None
+
+
+class EnrichIn(BaseModel):
+    source: str = Field(default="google_maps", description="google_maps | linkedin")
+    queries: List[str] = Field(..., min_length=1, max_length=20)
+    limit: int = Field(default=20, ge=1, le=500)
+
+
+class ScoreIn(BaseModel):
+    account_id: str
+
+
+class MemoryIn(BaseModel):
+    text: str = Field(..., min_length=1, max_length=8000)
+    metadata: Optional[Dict[str, Any]] = None
+
+
+class MemoryBatchIn(BaseModel):
+    texts: List[str] = Field(..., min_length=1, max_length=200)
+    metadata: Optional[Dict[str, Any]] = None
+
+
+class MemorySearchIn(BaseModel):
+    query: str = Field(..., min_length=1)
+    k: int = Field(default=8, ge=1, le=50)
+
+
+# ── health ───────────────────────────────────────────────────────────────────
+
+@app.get("/")
+async def root():
+    return {"app": settings.app_name, "version": settings.app_version, "docs": "/docs"}
+
+
+@app.get("/api/health")
+@app.get(f"{api_prefix}/health")
+@app.get("/health")
+async def health():
+    os_ok = await opensearch.store.ping()
+    apify_ok = await apify.ping()
+    cache_ok = await cache.ping()
+    components = {
+        "opensearch": {"ok": os_ok, "configured": opensearch.store.configured},
+        "apify": {"ok": apify_ok, "configured": apify.configured},
+        "graph": {"mode": graph.mode},
+        "cache": {"mode": cache.mode, "ok": cache_ok},
+        "embeddings": {"provider": provider_name()},
+        "enforcement": bool(settings.license_enforcement),
+    }
+    counts = {}
+    if os_ok:
+        try:
+            counts = {
+                "accounts": await opensearch.store.count("accounts"),
+                "contacts": await opensearch.store.count("contacts"),
+                "memory": await opensearch.store.count("memory"),
+            }
+        except Exception:
+            pass
+    core_ok = os_ok  # the store is the only hard dependency
+    return {
+        "status": "healthy" if core_ok else "degraded",
+        "app": settings.app_name,
+        "version": settings.app_version,
+        "components": components,
+        "counts": counts,
+        "timestamp": __import__("time").time(),
+    }
+
+
+# ── accounts ────────────────────────────────────────────────────────────────
+
+@app.post(f"{api_prefix}/accounts")
+async def create_account(payload: AccountIn):
+    try:
+        doc = await accounts_svc.upsert_account(payload.model_dump(exclude_none=True))
+    except ValueError as exc:
+        raise HTTPException(400, str(exc))
+    except Exception as exc:
+        raise HTTPException(503, f"Datastore unavailable: {exc}")
+    await graph.upsert_account(doc)
+    return {"account": doc}
+
+
+@app.get(f"{api_prefix}/accounts")
+async def list_accounts(q: str = "", industry: str = "", country: str = "",
+                        min_score: float = 0.0, limit: int = 25):
+    try:
+        hits = await accounts_svc.search_accounts(q, industry, country, min_score, limit)
+    except Exception as exc:
+        raise HTTPException(503, f"Datastore unavailable: {exc}")
+    return {"total": len(hits), "accounts": hits}
+
+
+@app.get(f"{api_prefix}/accounts/{{account_id}}")
+async def read_account(account_id: str):
+    result = await accounts_svc.get_account(account_id)
+    if result is None:
+        raise HTTPException(404, f"Account {account_id} not found")
+    return result
+
+
+# ── enrichment ──────────────────────────────────────────────────────────────
+
+@app.post(f"{api_prefix}/enrich")
+async def run_enrich(payload: EnrichIn):
+    try:
+        summary = await enrich_svc.enrich(payload.source, payload.queries, payload.limit)
+    except ValueError as exc:
+        raise HTTPException(400, str(exc))
+    except RuntimeError as exc:
+        raise HTTPException(502, str(exc))
+    except Exception as exc:
+        logger.exception("enrichment failed")
+        raise HTTPException(500, f"Enrichment failed: {exc}")
+    return summary
+
+
+# ── signals ─────────────────────────────────────────────────────────────────
+
+@app.post(f"{api_prefix}/signals/score")
+async def score(payload: ScoreIn):
+    account = await opensearch.store.get_doc("accounts", payload.account_id)
+    if account is None:
+        raise HTTPException(404, f"Account {payload.account_id} not found")
+    try:
+        res = await opensearch.store.client().post(
+            f"/{settings.os_index_prefix}-contacts/_count",
+            json={"query": {"term": {"account_id": payload.account_id}}},
+        )
+        contact_count = int(res.json().get("count", 0)) if res.status_code == 200 else 0
+    except Exception:
+        contact_count = 0
+    scored = signals_svc.score_account(account, contact_count)
+    doc = {**account, **scored}
+    await opensearch.store.upsert_doc("accounts", doc, payload.account_id)
+    await accounts_svc.record_event(payload.account_id, "rescored", scored)
+    return scored
+
+
+# ── vector memory ───────────────────────────────────────────────────────────
+
+@app.post(f"{api_prefix}/memory")
+async def memory_add(payload: MemoryIn):
+    try:
+        return await memory_svc.remember(payload.text, payload.metadata)
+    except ValueError as exc:
+        raise HTTPException(400, str(exc))
+    except Exception as exc:
+        raise HTTPException(503, f"Datastore unavailable: {exc}")
+
+
+@app.post(f"{api_prefix}/memory/batch")
+async def memory_add_batch(payload: MemoryBatchIn):
+    try:
+        return {"docs": await memory_svc.remember_batch(payload.texts, payload.metadata)}
+    except Exception as exc:
+        raise HTTPException(503, f"Datastore unavailable: {exc}")
+
+
+@app.post(f"{api_prefix}/memory/search")
+async def memory_search(payload: MemorySearchIn):
+    try:
+        hits = await memory_svc.recall(payload.query, payload.k)
+    except Exception as exc:
+        raise HTTPException(503, f"Datastore unavailable: {exc}")
+    return {"total": len(hits), "results": hits}
+
+
+# ── graph ───────────────────────────────────────────────────────────────────
+
+@app.get(f"{api_prefix}/graph/overview")
+async def graph_overview():
+    return await graph.overview()

app/services/accounts.py

@@ -0,0 +1,140 @@
+"""Account + contact services on top of OpenSearch."""
+import uuid
+from datetime import datetime, timezone
+from typing import Any, Dict, List, Optional
+
+from ..stores import opensearch
+
+
+def _now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+async def upsert_account(payload: Dict[str, Any]) -> Dict[str, Any]:
+    """Create or update an account keyed by domain (fallback: name hash)."""
+    domain = (payload.get("domain") or "").lower().strip() or None
+    name = (payload.get("name") or "").strip() or None
+    if not domain and not name:
+        raise ValueError("domain or name is required")
+
+    existing = None
+    if domain:
+        hits = await opensearch.store.search("accounts", {
+            "query": {"term": {"domain": domain}},
+            "size": 1,
+        })
+        existing = hits[0] if hits else None
+
+    account_id = (existing or {}).get("account_id") or str(uuid.uuid4())
+    doc = {
+        "account_id": account_id,
+        "domain": domain,
+        "name": name or (existing or {}).get("name"),
+        "industry": payload.get("industry") or (existing or {}).get("industry"),
+        "employee_count": payload.get("employee_count") or (existing or {}).get("employee_count"),
+        "hq_country": payload.get("hq_country") or (existing or {}).get("hq_country"),
+        "city": payload.get("city") or (existing or {}).get("city"),
+        "website": payload.get("website") or (existing or {}).get("website"),
+        "phone": payload.get("phone") or (existing or {}).get("phone"),
+        "address": payload.get("address") or (existing or {}).get("address"),
+        "description": payload.get("description") or (existing or {}).get("description"),
+        "rating": payload.get("rating") or (existing or {}).get("rating"),
+        "source": payload.get("source") or (existing or {}).get("source") or "manual",
+        "icp_score": (existing or {}).get("icp_score"),
+        "icp_label": (existing or {}).get("icp_label"),
+        "updated_at": _now(),
+    }
+    doc = {k: v for k, v in doc.items() if v is not None}
+    await opensearch.store.upsert_doc("accounts", doc, account_id)
+    return doc
+
+
+async def get_account(account_id: str) -> Optional[dict]:
+    account = await opensearch.store.get_doc("accounts", account_id)
+    if account is None:
+        return None
+    contacts = await opensearch.store.search("contacts", {
+        "query": {"term": {"account_id": account_id}},
+        "size": 100,
+        "sort": [{"_id": "asc"}],
+    })
+    events = await opensearch.store.search("events", {
+        "query": {"term": {"account_id": account_id}},
+        "size": 50,
+        "sort": [{"created_at": {"order": "desc"}}],
+    })
+    return {"account": account, "contacts": contacts, "events": events}
+
+
+async def search_accounts(q: str = "", industry: str = "", country: str = "",
+                          min_score: float = 0.0, limit: int = 25) -> List[dict]:
+    filters: List[dict] = []
+    if industry:
+        filters.append({"term": {"industry": industry.lower()}})
+    if country:
+        filters.append({"term": {"hq_country": country.lower()}})
+    if min_score > 0:
+        filters.append({"range": {"icp_score": {"gte": min_score}}})
+    must: List[dict] = []
+    if q:
+        must.append({"multi_match": {
+            "query": q,
+            "fields": ["name^3", "name.kw^2", "domain^3", "description", "industry"],
+            "fuzziness": "AUTO",
+        }})
+    body = {
+        "query": {"bool": {"must": must or [{"match_all": {}}], "filter": filters}},
+        "size": min(max(limit, 1), 100),
+    }
+    return await opensearch.store.search("accounts", body)
+
+
+async def upsert_contact(account_id: str, contact: Dict[str, Any]) -> Optional[dict]:
+    """Dedupe contacts per account on email (fallback: linkedin_url)."""
+    email = (contact.get("email") or "").lower().strip() or None
+    li = contact.get("linkedin_url")
+    existing = None
+    if email:
+        hits = await opensearch.store.search("contacts", {
+            "query": {"bool": {"filter": [
+                {"term": {"account_id": account_id}}, {"term": {"email": email}}
+            ]}},
+            "size": 1,
+        })
+        existing = hits[0] if hits else None
+    if existing is None and li:
+        hits = await opensearch.store.search("contacts", {
+            "query": {"bool": {"filter": [
+                {"term": {"account_id": account_id}}, {"term": {"linkedin_url": li}}
+            ]}},
+            "size": 1,
+        })
+        existing = hits[0] if hits else None
+
+    contact_id = (existing or {}).get("contact_id") or str(uuid.uuid4())
+    doc = {
+        "contact_id": contact_id,
+        "account_id": account_id,
+        "email": email,
+        "full_name": contact.get("full_name"),
+        "title": contact.get("title"),
+        "linkedin_url": li,
+        "phone": contact.get("phone"),
+        "source": contact.get("source") or "manual",
+        "updated_at": _now(),
+    }
+    doc = {k: v for k, v in doc.items() if v is not None}
+    await opensearch.store.upsert_doc("contacts", doc, contact_id)
+    return doc
+
+
+async def record_event(account_id: str, event_type: str, payload: dict) -> dict:
+    doc = {
+        "event_id": str(uuid.uuid4()),
+        "account_id": account_id,
+        "type": event_type,
+        "payload": payload,
+        "created_at": _now(),
+    }
+    await opensearch.store.index_doc("events", doc)
+    return doc

app/services/enrich.py

@@ -0,0 +1,159 @@
+"""Enrichment pipeline: Apify actors → normalized accounts/contacts →
+OpenSearch + ICP score + vector memory + optional Neo4j mirror."""
+import logging
+import uuid
+from datetime import datetime, timezone
+from typing import Any, Dict, List, Optional
+
+from ..config import settings
+from ..stores import opensearch
+from ..stores.apify import apify_client as apify, normalize_account, normalize_contact
+from ..stores.cache import cache
+from ..stores.graph import graph
+from . import accounts as accounts_svc
+from . import signals as signals_svc
+
+logger = logging.getLogger("abm.enrich")
+
+
+def _now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+SOURCE_ACTORS = {
+    "google_maps": lambda: settings.apify_actor_google_maps,
+    "linkedin": lambda: settings.apify_actor_linkedin,
+}
+
+
+def build_actor_input(source: str, queries: List[str], limit: int) -> dict:
+    """Actor input is schema-flexible on purpose: both of the owner's actors
+    accept a search-terms list plus result caps under slightly different keys,
+    so the pipeline sets the common superset and lets the actor ignore the
+    rest."""
+    if source == "linkedin":
+        return {
+            "searchUrl": queries[0] if queries and queries[0].startswith("http") else None,
+            "companies": [q for q in queries if not q.startswith("http")],
+            "maxItems": limit,
+            "maxResults": limit,
+            "scrapeEmails": True,
+        }
+    return {
+        "searchStringsArray": queries,
+        "maxCrawledPlaces": limit,
+        "maxCrawledPlacesPerSearch": max(limit // max(len(queries), 1), 1),
+        "language": "en",
+        "scrapeContacts": True,
+        "scrapeWebsiteEmails": True,
+    }
+
+
+async def enrich(
+    source: str,
+    queries: List[str],
+    limit: int = 20,
+    account_limit: int = 100,
+) -> Dict[str, Any]:
+    """Run one enrichment batch end-to-end. Returns a summary with counts."""
+    if source not in SOURCE_ACTORS:
+        raise ValueError(f"unknown source '{source}' (use google_maps or linkedin)")
+    if not apify.configured:
+        raise RuntimeError("Apify is not configured (APIFY_TOKEN missing)")
+    actor = SOURCE_ACTORS[source]()
+    run_input = build_actor_input(source, queries, limit)
+
+    cache_key = f"enrich:{source}:{hash(tuple(sorted(queries)))}:{limit}"
+    cached = await cache.get(cache_key)
+    if cached:
+        return {**cached, "cached": True}
+
+    started = _now()
+    items = await apify.run_sync(actor, run_input)
+    summary: Dict[str, Any] = {
+        "source": source,
+        "actor": actor,
+        "queries": queries,
+        "items_raw": len(items),
+        "accounts": 0,
+        "contacts": 0,
+        "skipped": 0,
+        "account_ids": [],
+        "started_at": started,
+        "finished_at": None,
+    }
+
+    if source == "linkedin":
+        # LinkedIn items are people; group them under a company account when
+        # the item names one, otherwise under the query itself.
+        for raw in items[:account_limit * 5]:
+            contact = normalize_contact(raw)
+            if not contact:
+                summary["skipped"] += 1
+                continue
+            company_name = contact.pop("company", None) or (queries[0] if queries else "Unknown")
+            domain = contact.get("email", "").split("@")[-1] if contact.get("email") else None
+            account = await accounts_svc.upsert_account({
+                "name": company_name, "domain": domain, "source": "linkedin",
+            })
+            saved = await accounts_svc.upsert_contact(account["account_id"], contact)
+            if saved:
+                summary["contacts"] += 1
+            await graph.upsert_contact(account["account_id"], saved or {})
+            if account["account_id"] not in summary["account_ids"]:
+                summary["account_ids"].append(account["account_id"])
+        summary["accounts"] = len(summary["account_ids"])
+    else:
+        for raw in items[:account_limit]:
+            account_fields = normalize_account(raw)
+            if not account_fields:
+                summary["skipped"] += 1
+                continue
+            account = await accounts_svc.upsert_account(account_fields)
+            summary["account_ids"].append(account["account_id"])
+            summary["accounts"] += 1
+            # Map-scraper items sometimes carry contact rows alongside.
+            raw_contacts = raw.get("emails") or raw.get("contacts") or []
+            if isinstance(raw_contacts, list):
+                for c in raw_contacts:
+                    if isinstance(c, str):
+                        c = {"email": c}
+                    contact = normalize_contact(c)
+                    if contact:
+                        saved = await accounts_svc.upsert_contact(account["account_id"], contact)
+                        if saved:
+                            summary["contacts"] += 1
+            await _finalize_account(account)
+            await graph.upsert_account(account)
+
+    summary["finished_at"] = _now()
+    await cache.set(cache_key, summary, ttl_s=900)
+    await opensearch.store.refresh("accounts")
+    await opensearch.store.refresh("contacts")
+    return summary
+
+
+async def _finalize_account(account: dict) -> None:
+    """Score + embed + persist enrichment events for one account."""
+    contacts = await opensearch.store.search("contacts", {
+        "query": {"term": {"account_id": account["account_id"]}},
+        "size": 1,
+    })
+    # count without pulling rows
+    from ..stores.opensearch import store
+    try:
+        res = await store.client().post(
+            f"/{store._index('contacts')}/_count",
+            json={"query": {"term": {"account_id": account["account_id"]}}},
+        )
+        contact_count = int(res.json().get("count", 0)) if res.status_code == 200 else 0
+    except Exception:
+        contact_count = len(contacts)
+
+    scored = signals_svc.score_account(account, contact_count)
+    doc = {**account, **scored, "embedding": signals_svc.embed_account(account)}
+    await store.upsert_doc("accounts", doc, account["account_id"])
+    await accounts_svc.record_event(account["account_id"], "enrichment_completed", {
+        "icp_score": scored["icp_score"], "icp_label": scored["icp_label"],
+        "contacts": contact_count, "at": _now(),
+    })

app/services/memory.py

@@ -0,0 +1,73 @@
+"""Vector memory over the OpenSearch k-NN index."""
+import json
+import uuid
+from datetime import datetime, timezone
+from typing import Any, Dict, List, Optional
+
+from ..stores import opensearch
+from ..stores.embeddings import embed, embed_batch, provider_name
+
+
+def _now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _memory_index() -> str:
+    return f"{opensearch.settings.os_index_prefix}-memory"
+
+
+async def remember(text: str, metadata: Optional[dict] = None) -> dict:
+    text = (text or "").strip()
+    if not text:
+        raise ValueError("text must not be empty")
+    doc_id = str(uuid.uuid4())
+    doc = {
+        "doc_id": doc_id,
+        "text": text[:8000],
+        "metadata": metadata or {},
+        "embedding": embed(text),
+        "created_at": _now(),
+    }
+    await opensearch.store.index_doc("memory", doc)
+    return {"doc_id": doc_id, "provider": provider_name()}
+
+
+async def remember_batch(texts: List[str], metadata: Optional[dict] = None) -> List[dict]:
+    texts = [t.strip()[:8000] for t in texts if t and t.strip()]
+    if not texts:
+        return []
+    vectors = embed_batch(texts)
+    docs = []
+    for text, vector in zip(texts, vectors):
+        docs.append({
+            "doc_id": str(uuid.uuid4()),
+            "text": text,
+            "metadata": metadata or {},
+            "embedding": vector,
+            "created_at": _now(),
+        })
+    lines: List[str] = []
+    for d in docs:
+        lines.append(json.dumps({"index": {"_index": _memory_index(), "_id": d["doc_id"]}}))
+        lines.append(json.dumps(d))
+    res = await opensearch.store.client().post(
+        "/_bulk",
+        content=("\n".join(lines) + "\n").encode(),
+        headers={"Content-Type": "application/x-ndjson"},
+    )
+    res.raise_for_status()
+    await opensearch.store.refresh("memory")
+    return [{"doc_id": d["doc_id"]} for d in docs]
+
+
+async def recall(query: str, k: int = 8) -> List[dict]:
+    query = (query or "").strip()
+    if not query:
+        return []
+    k = max(1, min(k, 50))
+    body = {
+        "size": k,
+        "query": {"knn": {"embedding": {"vector": embed(query), "k": k}}},
+        "_source": ["text", "metadata", "created_at"],
+    }
+    return await opensearch.store.search("memory", body)

app/services/signals.py

@@ -0,0 +1,68 @@
+"""ICP scoring — deterministic, explainable account prioritization.
+
+Score 0-100 from additive signals with the breakdown preserved, so sales can
+see WHY an account scored what it scored. Labels: A ≥ 75, B ≥ 50, C ≥ 30.
+"""
+from typing import Any, Dict
+
+from ..config import settings
+from ..stores import embeddings
+
+
+def score_account(account: Dict[str, Any], contact_count: int = 0) -> Dict[str, Any]:
+    breakdown: Dict[str, float] = {}
+
+    industry = str(account.get("industry") or "").lower()
+    hit = next((x for x in settings.icp_industries_list if x and x in industry), None)
+    breakdown["industry_fit"] = 20.0 if hit else 0.0
+
+    country = str(account.get("hq_country") or "").lower()
+    breakdown["geo_fit"] = 15.0 if any(x in country for x in settings.icp_countries_list if x) else 0.0
+
+    headcount = account.get("employee_count")
+    if isinstance(headcount, str) and headcount.isdigit():
+        headcount = int(headcount)
+    if isinstance(headcount, (int, float)) and headcount:
+        lo, hi = settings.icp_min_headcount, settings.icp_max_headcount
+        breakdown["size_band"] = 15.0 if lo <= headcount <= hi else 5.0
+    else:
+        breakdown["size_band"] = 0.0
+
+    breakdown["digital_presence"] = 15.0 if (account.get("domain") or account.get("website")) else 0.0
+
+    rating = account.get("rating")
+    if isinstance(rating, (int, float)) and rating:
+        # Google rating (1-5) → up to 10 points; presence alone is a weak signal
+        breakdown["market_signal"] = min(max((rating - 3.0) / 2.0, 0.0), 1.0) * 10.0
+    elif isinstance(rating, str) and rating.replace(".", "").isdigit():
+        breakdown["market_signal"] = min(max((float(rating) - 3.0) / 2.0, 0.0), 1.0) * 10.0
+    else:
+        breakdown["market_signal"] = 0.0
+
+    breakdown["contact_coverage"] = min(contact_count, 5) * 2.0  # up to 10
+
+    score = round(min(sum(breakdown.values()), 100.0), 1)
+    label = "A" if score >= 75 else "B" if score >= 50 else "C" if score >= 30 else "D"
+    return {
+        "icp_score": score,
+        "icp_label": label,
+        "breakdown": {k: round(v, 1) for k, v in breakdown.items()},
+        "scoring_version": "icp-v1",
+    }
+
+
+def account_text(account: Dict[str, Any]) -> str:
+    """The text that represents an account in the vector memory."""
+    parts = [
+        account.get("name") or "",
+        account.get("industry") or "",
+        account.get("description") or "",
+        f"domain {account.get('domain')}" if account.get("domain") else "",
+        f"city {account.get('city')}" if account.get("city") else "",
+        f"country {account.get('hq_country')}" if account.get("hq_country") else "",
+    ]
+    return ". ".join(p for p in parts if p)
+
+
+def embed_account(account: Dict[str, Any]):
+    return embeddings.embed(account_text(account))

app/stores/apify.py

@@ -0,0 +1,155 @@
+"""Apify enrichment client — runs the owner's own actors and fetches results.
+
+Actors (defaults are the account's own published actors, configurable via env):
+  - sitcod3.lab/fdn-google-map-master-scraper   (Google Maps → companies)
+  - sitcod3.lab/linkedin-company-email-scraper  (LinkedIn → employees + emails)
+
+Both sync (run-sync-get-dataset-items, capped) and async (run + poll) modes
+are supported. Output items are schema-flexible: the normalizer probes common
+Google Maps / LinkedIn export field names case-insensitively.
+"""
+import asyncio
+import logging
+import time
+from typing import Any, Dict, List, Optional
+
+import httpx
+
+from ..config import settings
+
+logger = logging.getLogger("abm.apify")
+
+API = "https://api.apify.com/v2"
+
+
+class ApifyClient:
+    def __init__(self) -> None:
+        self._http: Optional[httpx.AsyncClient] = None
+
+    @property
+    def configured(self) -> bool:
+        return settings.apify_configured
+
+    def client(self) -> httpx.AsyncClient:
+        if self._http is None:
+            self._http = httpx.AsyncClient(
+                base_url=API, params={"token": settings.apify_token},
+                timeout=httpx.Timeout(60.0),
+            )
+        return self._http
+
+    async def close(self) -> None:
+        if self._http is not None:
+            await self._http.aclose()
+            self._http = None
+
+    async def ping(self) -> bool:
+        if not self.configured:
+            return False
+        try:
+            res = await self.client().get("/users/me")
+            return res.status_code == 200
+        except Exception:
+            return False
+
+    async def run_sync(self, actor: str, run_input: dict, timeout_s: Optional[int] = None) -> List[dict]:
+        """Run an actor and collect its default dataset items (capped wait)."""
+        deadline = timeout_s or settings.apify_run_timeout_s
+        url = f"/acts/{actor}/run-sync-get-dataset-items"
+        params = {"timeout": int(deadline * 1000)}
+        res = await self.client().post(url, params=params, json=run_input)
+        if res.status_code in (402, 403):
+            raise RuntimeError(f"Apify rejected the run (HTTP {res.status_code}): {res.text[:200]}")
+        res.raise_for_status()
+        try:
+            items = res.json()
+            return items if isinstance(items, list) else []
+        except Exception:
+            return []
+
+    async def run_async(self, actor: str, run_input: dict) -> Dict[str, str]:
+        """Start an actor run; returns {run_id, dataset_id} for polling."""
+        res = await self.client().post(f"/acts/{actor}/runs", json=run_input)
+        res.raise_for_status()
+        data = res.json().get("data", {})
+        return {"run_id": data.get("id"), "dataset_id": data.get("defaultDatasetId")}
+
+    async def wait_run(self, run_id: str, poll_s: int = 10, timeout_s: Optional[int] = None) -> str:
+        deadline = time.monotonic() + (timeout_s or settings.apify_run_timeout_s)
+        while time.monotonic() < deadline:
+            res = await self.client().get(f"/actor-runs/{run_id}")
+            res.raise_for_status()
+            data = res.json().get("data", {})
+            status = data.get("status")
+            if status in ("SUCCEEDED", "FAILED", "ABORTED", "TIMED-OUT"):
+                return status
+            await asyncio.sleep(poll_s)
+        return "TIMED-OUT"
+
+    async def dataset_items(self, dataset_id: str, limit: int = 5000) -> List[dict]:
+        res = await self.client().get(f"/datasets/{dataset_id}/items", params={"limit": limit})
+        res.raise_for_status()
+        items = res.json()
+        return items if isinstance(items, list) else []
+
+
+apify_client = ApifyClient()
+
+
+# ── normalization: raw actor items → ABM records ─────────────────────────────
+
+def _pick(item: dict, *keys: str) -> Any:
+    low = {k.lower(): v for k, v in item.items() if isinstance(k, str)}
+    for k in keys:
+        if k in low and low[k]:
+            return low[k]
+    return None
+
+
+def normalize_account(raw: dict) -> Optional[dict]:
+    """Map a Google Maps-ish item to an ABM account (schema-flexible)."""
+    if not isinstance(raw, dict):
+        return None
+    name = _pick(raw, "name", "title", "companyname", "company", "searchstring")
+    if not name:
+        return None
+    website = _pick(raw, "website", "websiteurl", "url", "site", "domain")
+    domain = _pick(raw, "domain", "website_domain")
+    if not domain and website:
+        domain = str(website).replace("https://", "").replace("http://", "").split("/")[0]
+    if isinstance(domain, str):
+        domain = domain.removeprefix("www.").lower() or None
+    return {
+        "name": str(name)[:200],
+        "domain": domain,
+        "website": website,
+        "phone": _pick(raw, "phone", "phonenumber", "telephone"),
+        "industry": _pick(raw, "category", "maincategory", "categories", "industry", "subcategories"),
+        "city": _pick(raw, "city", "locality"),
+        "hq_country": _pick(raw, "country", "countrycode"),
+        "address": _pick(raw, "address", "fulladdress", "location", "street", "addressline"),
+        "employee_count": _pick(raw, "employeecount", "total_employees", "employees"),
+        "rating": _pick(raw, "rating", "totalScore", "reviews", "reviewCount"),
+        "description": _pick(raw, "description", "about", "snippet"),
+        "source": "google_maps",
+    }
+
+
+def normalize_contact(raw: dict, account_id_hint: Optional[str] = None) -> Optional[dict]:
+    """Map a LinkedIn-ish item to an ABM contact."""
+    if not isinstance(raw, dict):
+        return None
+    email = _pick(raw, "email", "emailaddress", "mail", "business_email")
+    full_name = _pick(raw, "fullname", "name", "full_name", " firstname") or \
+        " ".join(x for x in [_pick(raw, "firstname", "first_name"), _pick(raw, "lastname", "last_name")] if x)
+    if not email and not full_name:
+        return None
+    return {
+        "email": str(email).strip().lower() if email else None,
+        "full_name": str(full_name).strip()[:160] if full_name else None,
+        "title": _pick(raw, "title", "position", "jobtitle", "role", "headline"),
+        "linkedin_url": _pick(raw, "linkedinurl", "linkedin_url", "profile", "url"),
+        "phone": _pick(raw, "phone", "phonenumber"),
+        "company": _pick(raw, "company", "companyname", "organization", "current_company"),
+        "source": "linkedin",
+    }

app/stores/cache.py

@@ -0,0 +1,79 @@
+"""Optional Upstash Redis cache (REST) with an in-process TTL fallback.
+
+Upstash credentials arrive as REST URL + token (the console shows both). When
+they are absent the engine uses a process-local TTL dict — correct per
+container, lost on restart, which is acceptable for enrichment caching.
+"""
+import json
+import logging
+import time
+from typing import Any, Optional
+
+import httpx
+
+from ..config import settings
+
+logger = logging.getLogger("abm.cache")
+
+
+class Cache:
+    def __init__(self) -> None:
+        self._http: Optional[httpx.AsyncClient] = None
+        self._mem: dict[str, tuple[float, str]] = {}
+
+    @property
+    def mode(self) -> str:
+        return "upstash" if settings.upstash_configured else "memory"
+
+    def _client(self) -> httpx.AsyncClient:
+        if self._http is None:
+            self._http = httpx.AsyncClient(
+                base_url=settings.upstash_rest_url.rstrip("/"),
+                headers={"Authorization": f"Bearer {settings.upstash_rest_token}"},
+                timeout=httpx.Timeout(10.0),
+            )
+        return self._http
+
+    async def get(self, key: str) -> Optional[Any]:
+        if settings.upstash_configured:
+            try:
+                res = await self._client().get(f"/get/{key}")
+                if res.status_code == 200:
+                    raw = res.json().get("result")
+                    return json.loads(raw) if raw else None
+            except Exception as exc:
+                logger.warning("cache get failed: %s", exc)
+                return None
+        item = self._mem.get(key)
+        if not item:
+            return None
+        expires, raw = item
+        if expires < time.time():
+            self._mem.pop(key, None)
+            return None
+        return json.loads(raw)
+
+    async def set(self, key: str, value: Any, ttl_s: int = 3600) -> None:
+        raw = json.dumps(value, default=str)
+        if settings.upstash_configured:
+            try:
+                await self._client().get(f"/set/{key}?EX={int(ttl_s)}",
+                                         headers={"Content-Type": "text/plain"},
+                                         content=raw.encode())
+                return
+            except Exception as exc:
+                logger.warning("cache set failed: %s", exc)
+                return
+        self._mem[key] = (time.time() + ttl_s, raw)
+
+    async def ping(self) -> bool:
+        if not settings.upstash_configured:
+            return True  # memory mode always "works"
+        try:
+            res = await self._client().get("/ping")
+            return res.status_code == 200
+        except Exception:
+            return False
+
+
+cache = Cache()

app/stores/embeddings.py

@@ -0,0 +1,114 @@
+"""Embedding providers for the ABM vector memory.
+
+`auto` prefers fastembed (ONNX MiniLM, 384-dim, real semantic similarity) and
+falls back to a deterministic hashing embedding when the model is unavailable
+(cold boot, build without the model cache). Both produce L2-normalized vectors
+of settings.embedding_dim so the k-NN index stays compatible either way.
+"""
+import hashlib
+import logging
+import math
+import re
+from typing import List
+
+from ..config import settings
+
+logger = logging.getLogger("abm.embeddings")
+
+_TOKEN_RE = re.compile(r"[a-z0-9]{2,}")
+
+
+class HashEmbedding:
+    """Deterministic feature-hashing embedding (bag of tokens + bigrams).
+
+    Honest about what it is: lexical-ish similarity, not deep semantics.
+    Zero dependencies, instant, stable across restarts — the safe default.
+    """
+
+    name = "hash-v1"
+
+    def _dims(self) -> int:
+        return settings.embedding_dim
+
+    def _index(self, token: str) -> tuple[int, float]:
+        digest = hashlib.md5(token.encode("utf-8")).digest()
+        idx = int.from_bytes(digest[:4], "little") % self._dims()
+        sign = 1.0 if digest[4] % 2 == 0 else -1.0
+        return idx, sign
+
+    def encode(self, text: str) -> List[float]:
+        vec = [0.0] * self._dims()
+        tokens = _TOKEN_RE.findall((text or "").lower())
+        grams: List[str] = list(tokens)
+        grams += [f"{a}_{b}" for a, b in zip(tokens, tokens[1:])]
+        for g in grams:
+            idx, sign = self._index(g)
+            vec[idx] += sign
+        norm = math.sqrt(sum(v * v for v in vec)) or 1.0
+        return [round(v / norm, 6) for v in vec]
+
+    def encode_batch(self, texts: List[str]) -> List[List[float]]:
+        return [self.encode(t) for t in texts]
+
+
+class FastEmbedEmbedding:
+    """ONNX MiniLM via fastembed — real semantic similarity, no torch."""
+
+    def __init__(self) -> None:
+        from fastembed import TextEmbedding  # imported lazily
+
+        self._model = TextEmbedding("sentence-transformers/all-MiniLM-L6-v2")
+        self.name = "fastembed-minilm-384"
+
+    def _normalize(self, vec) -> List[float]:
+        norm = math.sqrt(sum(v * v for v in vec)) or 1.0
+        return [round(float(v) / norm, 6) for v in vec]
+
+    def encode(self, text: str) -> List[float]:
+        out = list(self._model.embed([text]))[0]
+        return self._normalize(out)
+
+    def encode_batch(self, texts: List[str]) -> List[List[float]]:
+        return [self._normalize(v) for v in self._model.embed(texts)]
+
+
+_provider = None
+_tried = False
+
+
+def get_provider():
+    global _provider, _tried
+    if _provider is not None:
+        return _provider
+    if _tried:
+        return None
+    _tried = True
+    choice = (settings.embedding_provider or "auto").lower()
+    if choice in ("auto", "fastembed"):
+        try:
+            _provider = FastEmbedEmbedding()
+            logger.info("embeddings: fastembed active (%s)", _provider.name)
+            return _provider
+        except Exception as exc:
+            logger.warning("fastembed unavailable (%s) — using hash embeddings", exc)
+    _provider = HashEmbedding()
+    return _provider
+
+
+def embed(text: str) -> List[float]:
+    p = get_provider()
+    if p is None:
+        return HashEmbedding().encode(text)
+    return p.encode(text)
+
+
+def embed_batch(texts: List[str]) -> List[List[float]]:
+    p = get_provider()
+    if p is None:
+        return HashEmbedding().encode_batch(texts)
+    return p.encode_batch(texts)
+
+
+def provider_name() -> str:
+    p = get_provider()
+    return p.name if p else "hash-v1"

app/stores/graph.py

@@ -0,0 +1,182 @@
+"""Optional graph mirror — BetterDB (managed Neo4j) or direct Aura.
+
+The ABM graph (accounts, contacts, enrichment events) mirrors into Neo4j when
+a connection is available; the engine runs fully without it (OpenSearch is the
+source of truth). Two transports, auto-detected at boot:
+
+1. BETTERDB_URL + BETTERDB_TOKEN — managed Neo4j behind betterdb.com. The
+   HTTP query API is probed at a few known paths; BetterDB answers "waiting"
+   until the owner connects their instance in the console, in which case the
+   mirror stays disabled and /health reports it honestly.
+2. NEO4J_URI (+user/password) — a direct Neo4j/Aura endpoint via the official
+   driver (bolt+ssc for Aura).
+"""
+import logging
+from datetime import datetime, timezone
+from typing import Any, Dict, List, Optional
+
+import httpx
+
+from ..config import settings
+
+logger = logging.getLogger("abm.graph")
+
+
+class GraphStore:
+    def __init__(self) -> None:
+        self._mode: Optional[str] = None  # "betterdb" | "neo4j" | None
+        self._driver = None
+        self._http: Optional[httpx.AsyncClient] = None
+        self._query_path: Optional[str] = None
+
+    @property
+    def mode(self) -> Optional[str]:
+        return self._mode
+
+    async def connect(self) -> bool:
+        if settings.neo4j_uri:
+            try:
+                from neo4j import GraphDatabase  # optional dependency
+
+                scheme = "neo4j+ssc" if "+ssc" not in settings.neo4j_uri else None
+                uri = settings.neo4j_uri
+                if uri.startswith("neo4j://"):
+                    uri = uri.replace("neo4j://", "neo4j+ssc://", 1)
+                elif scheme and not uri.startswith("neo4j+"):
+                    uri = f"{scheme}://{uri}"
+                self._driver = GraphDatabase.driver(
+                    uri, auth=(settings.neo4j_user or "neo4j", settings.neo4j_password or "")
+                )
+                await asyncio_wrap(lambda: self._driver.verify_connectivity())
+                self._mode = "neo4j"
+                logger.info("graph: direct Neo4j connected (%s)", uri)
+                return True
+            except Exception as exc:
+                logger.warning("graph: direct Neo4j connect failed: %s", exc)
+                self._driver = None
+        if settings.betterdb_url and settings.betterdb_token:
+            self._http = httpx.AsyncClient(
+                base_url=settings.betterdb_url.rstrip("/"),
+                headers={"Authorization": f"Bearer {settings.betterdb_token}"},
+                timeout=httpx.Timeout(20.0),
+            )
+            for path in (
+                "/api/neo4j/query/v2",
+                "/neo4j/query/v2",
+                "/api/db/neo4j/query/v2",
+                "/db/neo4j/query/v2",
+                "/api/query",
+            ):
+                try:
+                    res = await self._http.post(
+                        path, json={"statement": "RETURN 1 AS ok"}
+                    )
+                    if res.status_code == 200:
+                        self._query_path = path
+                        self._mode = "betterdb"
+                        logger.info("graph: BetterDB query API live at %s", path)
+                        return True
+                    if res.status_code in (401, 402, 403):
+                        logger.warning("graph: BetterDB answered %s at %s — token %s",
+                                       res.status_code, path, "rejected")
+                except Exception:
+                    continue
+            logger.info("graph: BetterDB reachable but no query API answered yet "
+                        "(owner must connect the instance in the console)")
+        return False
+
+    async def close(self) -> None:
+        if self._http is not None:
+            await self._http.aclose()
+        if self._driver is not None:
+            self._driver.close()
+
+    async def run(self, cypher: str, params: Optional[Dict[str, Any]] = None) -> List[Dict[str, Any]]:
+        """Run a write/query statement; returns rows (best-effort, logged)."""
+        if self._mode == "betterdb" and self._query_path:
+            try:
+                res = await self._http.post(
+                    self._query_path, json={"statement": cypher, "parameters": params or {}}
+                )
+                if res.status_code != 200:
+                    logger.warning("graph query %s: %s", res.status_code, res.text[:150])
+                    return []
+                body = res.json()
+                return body.get("records") or body.get("data") or []
+            except Exception as exc:
+                logger.warning("graph query failed: %s", exc)
+                return []
+        if self._mode == "neo4j" and self._driver is not None:
+            def _run():
+                with self._driver.session() as session:
+                    result = session.run(cypher, params or {})
+                    return [r.data() for r in result]
+            try:
+                return await asyncio_wrap(_run)
+            except Exception as exc:
+                logger.warning("graph query failed: %s", exc)
+                return []
+        return []
+
+    # ── mirror API (idempotent MERGEs) ───────────────────────────────────
+
+    async def upsert_account(self, account: Dict[str, Any]) -> None:
+        if self._mode is None:
+            return
+        await self.run(
+            """
+            MERGE (a:Account {account_id: $account_id})
+            SET a.name = $name, a.domain = $domain, a.industry = $industry,
+                a.hq_country = $hq_country, a.icp_score = $icp_score,
+                a.updated_at = $updated_at
+            """,
+            {
+                "account_id": account.get("account_id"),
+                "name": account.get("name"),
+                "domain": account.get("domain"),
+                "industry": account.get("industry"),
+                "hq_country": account.get("hq_country"),
+                "icp_score": account.get("icp_score"),
+                "updated_at": datetime.now(timezone.utc).isoformat(),
+            },
+        )
+
+    async def upsert_contact(self, account_id: str, contact: Dict[str, Any]) -> None:
+        if self._mode is None:
+            return
+        await self.run(
+            """
+            MATCH (a:Account {account_id: $account_id})
+            MERGE (c:Contact {contact_id: $contact_id})
+            SET c.full_name = $full_name, c.title = $title, c.email = $email,
+                c.linkedin_url = $linkedin_url
+            MERGE (c)-[:WORKS_AT]->(a)
+            """,
+            {
+                "account_id": account_id,
+                "contact_id": contact.get("contact_id"),
+                "full_name": contact.get("full_name"),
+                "title": contact.get("title"),
+                "email": contact.get("email"),
+                "linkedin_url": contact.get("linkedin_url"),
+            },
+        )
+
+    async def overview(self) -> Dict[str, Any]:
+        if self._mode is None:
+            return {"mode": None, "status": "not connected"}
+        rows = await self.run(
+            "MATCH (a:Account) WITH count(a) AS accounts "
+            "OPTIONAL MATCH (c:Contact) RETURN accounts, count(c) AS contacts"
+        )
+        return {"mode": self._mode, "status": "connected", "counts": rows or []}
+
+
+import asyncio as _asyncio
+
+
+async def asyncio_wrap(fn):
+    return await _asyncio.to_thread(fn)
+
+
+graph = GraphStore()

app/stores/opensearch.py

@@ -0,0 +1,183 @@
+"""OpenSearch persistence layer — the ABM engine's primary store.
+
+Aiven OpenSearch (v3.6, k-NN plugin enabled, cluster green) carries accounts,
+contacts, signal events and the vector memory index. Everything degrades
+gracefully: if OpenSearch is unreachable the service reports unhealthy in
+/health instead of crashing, so the Space can boot before the datastore.
+"""
+import asyncio
+import logging
+import uuid
+from datetime import datetime, timezone
+from typing import Any, Dict, List, Optional
+
+import httpx
+
+from ..config import settings
+
+logger = logging.getLogger("abm.os")
+
+
+def _now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+class OpenSearchStore:
+    def __init__(self) -> None:
+        self._client: Optional[httpx.AsyncClient] = None
+
+    @property
+    def configured(self) -> bool:
+        return settings.os_configured
+
+    def client(self) -> httpx.AsyncClient:
+        if self._client is None:
+            auth = (settings.os_user, settings.os_password) if settings.os_user else None
+            self._client = httpx.AsyncClient(
+                base_url=settings.os_url.rstrip("/"),
+                auth=auth,
+                timeout=httpx.Timeout(30.0),
+                verify=True,
+            )
+        return self._client
+
+    async def close(self) -> None:
+        if self._client is not None:
+            await self._client.aclose()
+            self._client = None
+
+    # ── index bootstrap ──────────────────────────────────────────────────
+
+    def _index(self, name: str) -> str:
+        return f"{settings.os_index_prefix}-{name}"
+
+    async def _put_index(self, name: str, body: dict) -> bool:
+        res = await self.client().put(f"/{self._index(name)}", json=body)
+        if res.status_code in (200, 201):
+            return True
+        if res.status_code == 400 and "resource_already_exists" in res.text:
+            return True
+        logger.warning("put_index %s failed: %s %s", name, res.status_code, res.text[:200])
+        return False
+
+    async def bootstrap(self) -> dict:
+        """Create the four indices if missing. Returns per-index status."""
+        dim = settings.embedding_dim
+        knn_vector = {
+            "type": "knn_vector",
+            "dimension": dim,
+            "method": {"name": "hnsw", "space_type": "cosinesimil", "engine": "nmslib"},
+        }
+        results: Dict[str, bool] = {}
+        results["accounts"] = await self._put_index("accounts", {
+            "settings": {"index": {"knn": True}},
+            "mappings": {"properties": {
+                "account_id": {"type": "keyword"},
+                "domain": {"type": "keyword"},
+                "name": {"type": "text", "fields": {"kw": {"type": "keyword"}}},
+                "industry": {"type": "keyword"},
+                "employee_count": {"type": "integer"},
+                "hq_country": {"type": "keyword"},
+                "city": {"type": "keyword"},
+                "website": {"type": "keyword"},
+                "phone": {"type": "keyword"},
+                "description": {"type": "text"},
+                "source": {"type": "keyword"},
+                "icp_score": {"type": "float"},
+                "icp_label": {"type": "keyword"},
+                "signals": {"type": "flat_object"},
+                "embedding": knn_vector,
+                "updated_at": {"type": "date"},
+            }},
+        })
+        results["contacts"] = await self._put_index("contacts", {
+            "mappings": {"properties": {
+                "contact_id": {"type": "keyword"},
+                "account_id": {"type": "keyword"},
+                "email": {"type": "keyword"},
+                "full_name": {"type": "text"},
+                "title": {"type": "keyword"},
+                "linkedin_url": {"type": "keyword"},
+                "phone": {"type": "keyword"},
+                "source": {"type": "keyword"},
+                "updated_at": {"type": "date"},
+            }},
+        })
+        results["events"] = await self._put_index("events", {
+            "mappings": {"properties": {
+                "event_id": {"type": "keyword"},
+                "account_id": {"type": "keyword"},
+                "type": {"type": "keyword"},
+                "payload": {"type": "flat_object"},
+                "created_at": {"type": "date"},
+            }},
+        })
+        results["memory"] = await self._put_index("memory", {
+            "settings": {"index": {"knn": True}},
+            "mappings": {"properties": {
+                "doc_id": {"type": "keyword"},
+                "text": {"type": "text"},
+                "metadata": {"type": "flat_object"},
+                "embedding": knn_vector,
+                "created_at": {"type": "date"},
+            }},
+        })
+        return results
+
+    # ── generic helpers ────────────────────────────────��─────────────────
+
+    async def index_doc(self, index: str, doc: dict, doc_id: Optional[str] = None) -> str:
+        _id = doc_id or doc.get("account_id") or doc.get("contact_id") or doc.get("event_id") or doc.get("doc_id") or str(uuid.uuid4())
+        res = await self.client().put(f"/{self._index(index)}/_doc/{_id}", json=doc)
+        res.raise_for_status()
+        return _id
+
+    async def upsert_doc(self, index: str, doc: dict, doc_id: str) -> str:
+        res = await self.client().put(
+            f"/{self._index(index)}/_doc/{_id}", json=doc,
+            params={"refresh": "false"},
+        )
+        res.raise_for_status()
+        return doc_id
+
+    async def get_doc(self, index: str, doc_id: str) -> Optional[dict]:
+        res = await self.client().get(f"/{self._index(index)}/_doc/{doc_id}")
+        if res.status_code == 404:
+            return None
+        res.raise_for_status()
+        body = res.json()
+        if not body.get("found"):
+            return None
+        return {"_id": body["_id"], **body["_source"]}
+
+    async def delete_doc(self, index: str, doc_id: str) -> bool:
+        res = await self.client().delete(f"/{self._index(index)}/_doc/{doc_id}")
+        return res.status_code in (200, 404)
+
+    async def search(self, index: str, body: dict) -> List[dict]:
+        res = await self.client().post(f"/{self._index(index)}/_search", json=body)
+        res.raise_for_status()
+        hits = res.json().get("hits", {}).get("hits", [])
+        return [{"_id": h["_id"], "_score": h.get("_score"), **h["_source"]} for h in hits]
+
+    async def refresh(self, index: str) -> None:
+        try:
+            await self.client().post(f"/{self._index(index)}/_refresh")
+        except Exception:
+            pass
+
+    async def count(self, index: str) -> int:
+        res = await self.client().get(f"/{self._index(index)}/_count")
+        if res.status_code != 200:
+            return -1
+        return int(res.json().get("count", -1))
+
+    async def ping(self) -> bool:
+        try:
+            res = await self.client().get("/")
+            return res.status_code == 200
+        except Exception:
+            return False
+
+
+store = OpenSearchStore()

pytest.ini

@@ -0,0 +1,3 @@
+[pytest]
+testpaths = tests
+asyncio_mode = auto

requirements.txt

@@ -0,0 +1,19 @@
+# --- Core web framework ---
+fastapi>=0.115.0
+uvicorn[standard]>=0.30.6
+
+# --- Data validation ---
+pydantic>=2.11.0
+pydantic-settings>=2.7.0
+
+# --- HTTP ---
+httpx>=0.27.2
+
+# --- Auth (Supabase JWKS) ---
+python-jose[cryptography]>=3.3.0
+
+# --- Embeddings (optional at runtime; hash fallback always available) ---
+fastembed>=0.3.6
+
+# --- Optional direct Neo4j (Aura) ---
+neo4j>=5.25.0

tests/test_abm.py

@@ -0,0 +1,128 @@
+"""Unit tests — normalization, scoring, embeddings, auth, API contract."""
+import unittest
+from unittest.mock import AsyncMock, MagicMock, patch
+
+from fastapi.testclient import TestClient
+
+from app.main import app
+from app.stores.apify import normalize_account, normalize_contact
+from app.services.signals import score_account
+
+client = TestClient(app)
+
+
+class NormalizerTests(unittest.TestCase):
+    """Actor output is schema-flexible; the normalizer must survive it."""
+
+    def test_google_maps_item_full(self):
+        raw = {
+            "title": "Acme Robotics",
+            "website": "https://www.acmerobotics.com/contact",
+            "phoneNumber": "+1 555 0100",
+            "category": "Industrial equipment supplier",
+            "city": "Detroit",
+            "countryCode": "US",
+            "totalScore": 4.6,
+            "description": "Industrial robots",
+        }
+        out = normalize_account(raw)
+        self.assertEqual(out["name"], "Acme Robotics")
+        self.assertEqual(out["domain"], "acmerobotics.com")
+        self.assertEqual(out["phone"], "+1 555 0100")
+        self.assertEqual(out["source"], "google_maps")
+
+    def test_google_maps_item_missing_name_is_skipped(self):
+        self.assertIsNone(normalize_account({"website": "https://x.com"}))
+
+    def test_linkedin_contact(self):
+        raw = {
+            "fullName": "Jane Doe",
+            "email": "Jane.Doe@Example.com ",
+            "position": "Head of Ops",
+            "linkedinUrl": "https://linkedin.com/in/janedoe",
+            "company": "Acme",
+        }
+        out = normalize_contact(raw)
+        self.assertEqual(out["email"], "jane.doe@example.com")
+        self.assertEqual(out["title"], "Head of Ops")
+
+    def test_contact_requires_email_or_name(self):
+        self.assertIsNone(normalize_contact({"title": "CEO"}))
+
+
+class ScoringTests(unittest.TestCase):
+    def test_full_fit_account_scores_A(self):
+        account = {
+            "name": "Great Co", "domain": "great.com", "website": "great.com",
+            "industry": "Software", "hq_country": "United States",
+            "employee_count": 120, "rating": 4.8,
+        }
+        out = score_account(account, contact_count=5)
+        self.assertGreaterEqual(out["icp_score"], 75)
+        self.assertEqual(out["icp_label"], "A")
+        self.assertIn("industry_fit", out["breakdown"])
+
+    def test_poor_fit_account_scores_D(self):
+        out = score_account({"name": "Nobody"}, contact_count=0)
+        self.assertLess(out["icp_score"], 30)
+        self.assertEqual(out["icp_label"], "D")
+
+    def test_score_is_explainable_and_bounded(self):
+        out = score_account({"name": "X", "industry": "software", "hq_country": "usa",
+                             "employee_count": 50, "rating": 5.0}, contact_count=99)
+        self.assertLessEqual(out["icp_score"], 100)
+        self.assertEqual(out["breakdown"]["contact_coverage"], 10.0)
+
+
+class EmbeddingTests(unittest.TestCase):
+    def test_hash_embedding_deterministic_and_normalized(self):
+        from app.stores.embeddings import HashEmbedding
+        e1 = HashEmbedding().encode("acme robotics detroit")
+        e2 = HashEmbedding().encode("acme robotics detroit")
+        e3 = HashEmbedding().encode("completely different text")
+        self.assertEqual(e1, e2)
+        self.assertNotEqual(e1, e3)
+        self.assertEqual(len(e1), 384)
+        norm = sum(v * v for v in e1) ** 0.5
+        self.assertAlmostEqual(norm, 1.0, places=2)
+
+
+class AuthTests(unittest.TestCase):
+    def test_api_key_auth(self):
+        from app.config import settings
+        with patch.object(settings, "api_key", "test-key-123"):
+            from app.auth import authenticate
+            req = MagicMock()
+            req.headers = {"x-api-key": "test-key-123"}
+            import asyncio
+            out = asyncio.run(authenticate(req))
+            self.assertEqual(out["via"], "api-key")
+            self.assertEqual(out["user_id"], "service")
+
+
+class HealthTests(unittest.TestCase):
+    def test_health_is_open_and_reports_components(self):
+        res = client.get("/health")
+        self.assertEqual(res.status_code, 200)
+        body = res.json()
+        self.assertIn("components", body)
+        self.assertIn("embeddings", body["components"])
+        self.assertIn("status", body)
+
+
+class ApiContractTests(unittest.TestCase):
+    def test_enrich_rejects_unknown_source(self):
+        res = client.post("/api/v1/enrich", json={"source": "twitter", "queries": ["x"]})
+        self.assertEqual(res.status_code, 400)
+
+    def test_accounts_require_domain_or_name(self):
+        res = client.post("/api/v1/accounts", json={"industry": "software"})
+        self.assertEqual(res.status_code, 400)
+
+    def test_memory_rejects_empty_text(self):
+        res = client.post("/api/v1/memory", json={"text": "  "})
+        self.assertEqual(res.status_code, 400)
+
+
+if __name__ == "__main__":
+    unittest.main()