#!/usr/bin/env python
"""The lead stories, end to end, on a real DataK3 bucket (erp_platform.livecheck).
BUCKET=<scratch bucket> DODIL_ORG=<org> python scripts/live_check.py
What SQLite cannot show: the import's bulk upsert and the derived key on DataK3's ON CONFLICT, the score run
twice, the VECTOR(31) column, and pgvector's `<=>` ranking IN the database (SQLite ranks in Python). The
seams (convert) are not exercised here — they need the neighbours deployed beside this brick.
0.2.0 adds: the optimistic version on DataK3 (the migration adds the column and the claim table) — a stale one is
409, four edits racing from one version through a REAL engine give one success, and a re-import claims each lead's
next version in bulk; a CSV row that cannot be read is 422 with its row number.
"""
import pathlib
import sys
import threading
ROOT = pathlib.Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "src"))
from erp_platform import livecheck as lc # noqa: E402
from brick_lead.application import use_cases as uc # noqa: E402
from brick_lead.domain.lead import cosine_distance, lead_id_for, profile # noqa: E402
from brick_lead.infrastructure.tables import LeadRepository, md # noqa: E402
BIG = 54848654143250435 # a real-shaped bp_id: above 2^53
def person(email, roles, permissions):
return {"sub": email, "email": email, "roles": roles, "permissions": permissions, "auth_source": "pool"}
TOM = person("tom@co.example", ["sales"], ["lead:lead:write"])
SOFIA = person("sofia@co.example", ["sales"], ["lead:lead:write"])
BOSS = person("max@co.example", ["manager"], ["lead:lead:write", "lead:lead:assign", "lead:policy:write"])
MIA = person("mia@co.example", ["marketing"], ["lead:lead:import"])
JOB = person("scoring@svc", ["service"], ["lead:lead:score"])
HOOK = person("campaign@svc", ["service"], ["lead:signal:record"])
LIST = """Name,Email,Company,Website,Country,Industry,Size,Owner
Ana Berg,ana@samesoft.se,Samesoft,samesoft.se,SE,software,51-200,tom@co.example
Bo Ek,bo@nordsoft.no,Nordsoft,nordsoft.no,NO,SaaS,250,tom@co.example
Cy Mart,cy@bigretail.com,Big Retail,bigretail.com,US,retail,12000,tom@co.example
No Email,,Nobody,,,,,
Di Sofia,di@sofiasoft.se,Sofiasoft,sofiasoft.se,SE,software,51-200,sofia@co.example
"""
def main() -> int:
tables = [t.name for t in md.sorted_tables]
lc.create_schema(ROOT / "src" / "brick_lead" / "schema.sql", tables)
uc.repository = lc.repository(LeadRepository)
# ── CRM-13: import twice ─────────────────────────────────────────────────────────────
first = uc.import_leads(MIA, LIST, "Nordic Expo")
lc.check("CRM-13 first import: 4 created, the row without email reported as row 5",
(first["created"], first["skipped"]) == (4, [{"row": 5, "reason": "no email"}]), first)
ana = lead_id_for("ana@samesoft.se")
uc.start_working(TOM, ana)
again = uc.import_leads(MIA, LIST.replace("Samesoft,samesoft.se", "Samesoft AB,samesoft.se"), "another list")
stored = lc.rows("SELECT lead_id, owner, status, source, company FROM lead_leads ORDER BY lead_id")
lc.check("CRM-13 second import: nothing created, one updated, 4 rows in the table",
(again["created"], again["updated"], again["unchanged"], len(stored)) == (0, 1, 3, 4), (again, stored))
a = [r for r in stored if r["lead_id"] == ana][0]
lc.check("a re-import keeps the owner, the status and the first source; it takes the new company name",
(a["owner"], a["status"], a["source"], a["company"]) ==
("tom@co.example", "working", "nordic expo", "Samesoft AB"), a)
lc.check("CRM-01 a rep cannot see another rep's lead",
(lc.refused(uc.get_lead, SOFIA, ana) or (0,))[0] == 404)
# ── CRM-10: signals, scoring twice ───────────────────────────────────────────────────
as_of = "2026-10-17T12:00:00+00:00"
for kind, at in (("email_opened_pricing", "2026-10-17T11:00:00Z"), ("email_opened_pricing", "2026-10-17T11:00:00Z"),
("demo_requested", "2026-10-17T10:00:00+02:00"), ("webinar_attended", "2026-08-01T10:00:00Z")):
uc.record_signal(HOOK, ana, kind, at)
signals = lc.rows(f"SELECT kind FROM lead_signals WHERE lead_id = '{ana}' ORDER BY kind")
lc.check("the same event reported twice is one signal", [s["kind"] for s in signals] ==
["demo_requested", "email_opened_pricing", "webinar_attended"], signals)
gate = lc.refused(uc.qualify_lead, TOM, lead_id_for("bo@nordsoft.no"))
lc.check("qualify needs the lead to be worked first", gate and gate[0] == 409, gate)
run1 = uc.score_leads(JOB, as_of)
run2 = uc.score_leads(JOB, as_of)
got = uc.get_lead(TOM, ana)
lc.check("CRM-10 the run scored 4, qualified the worked lead at 10+25+40 = 75 (the August webinar is "
"outside the 30-day window)",
(run1["leads_scored"], run1["scores_changed"], run1["qualified"], got["score"], got["status"]) ==
(4, 4, [ana], 75, "qualified"), (run1, got))
lc.check("CRM-10 running it twice changes nothing: 0 changed, and the score keeps the run that computed it",
(run2["scores_changed"], run2["qualified"], got["scored_by_run"]) == (0, [], run1["run_id"]), run2)
runs = lc.rows("SELECT run_id, scores_changed FROM lead_score_runs ORDER BY scores_changed DESC")
lc.check("each run is recorded", [(r["run_id"], int(r["scores_changed"])) for r in runs] ==
[(run1["run_id"], 4), (run2["run_id"], 0)], runs)
# ── the workflow ─────────────────────────────────────────────────────────────────────
bo = lead_id_for("bo@nordsoft.no")
no_reason = lc.refused(uc.disqualify_lead, TOM, bo, " ")
uc.disqualify_lead(TOM, bo, "no budget this year")
lc.check("disqualifying needs a reason, then the lead is disqualified",
no_reason and no_reason[0] == 422 and uc.get_lead(TOM, bo)["status"] == "disqualified", no_reason)
lc.check("a disqualified lead is not re-scored", uc.score_leads(JOB, as_of)["leads_scored"] == 3)
# ── CRM-14: the VECTOR column and <=> on DataK3 ──────────────────────────────────────
vec = lc.rows(f"SELECT CAST(profile AS VARCHAR) AS p FROM lead_profiles WHERE lead_id = '{ana}'")
import json
stored_vec = json.loads(vec[0]["p"]) if vec else []
lc.check("the VECTOR(31) column holds the profile", len(stored_vec) == 31 and max(
abs(x - y) for x, y in zip(stored_vec, profile("software", "51-200", "SE"))) < 1e-5, vec)
with uc.repository() as repo: # a won account, converted from a lead
won = repo.get(lead_id_for("di@sofiasoft.se"))
won.status, won.bp_id, won.contact_id, won.opportunity_id = "converted", BIG, "C-LIVE", "O-LIVE"
repo.save(won, "live-check")
lc.check("bp_id above 2^53 round-trips exactly",
lc.rows(f"SELECT CAST(bp_id AS VARCHAR) AS b FROM lead_leads WHERE bp_id = {BIG}") == [{"b": str(BIG)}])
with uc.repository() as repo:
ranked = repo.nearest(profile("software", "51-200", "SE"), None, 10)
lc.check("<=> ranks in the database, nearest first, and agrees with the cosine distance computed here",
[l.domain for l, _ in ranked] == ["samesoft.se", "bigretail.com"] # Bo is disqualified, Di converted
and ranked[0][1] < ranked[1][1] and all(
abs(d - cosine_distance(profile(l.industry, l.size_band, l.country),
profile("software", "51-200", "SE"))) < 1e-5 for l, d in ranked)
and ranked[0][1] < 1e-5, [(l.domain, d) for l, d in ranked])
looks = uc.find_lookalikes(TOM, bp_id=BIG)
lc.check("CRM-14 lookalikes of the won account: its lead's profile, ranked, the customer itself excluded",
looks["based_on"]["from"] == f"lead {won.lead_id}"
and [c["domain"] for c in looks["lookalikes"]] == ["samesoft.se", "bigretail.com"], looks)
lc.check("CRM-14 a rep sees only their own candidates",
[c["domain"] for c in uc.find_lookalikes(SOFIA, industry="software", country="SE")["lookalikes"]] == [])
# ── configuration ────────────────────────────────────────────────────────────────────
uc.set_scoring_policy(BOSS, 20, 30, [{"attribute": "industry", "value": "retail", "points": 25}], [])
uc.score_leads(JOB, as_of)
lc.check("the scoring policy is rows: a new policy changes the next run's scores",
uc.get_lead(TOM, lead_id_for("cy@bigretail.com"))["score"] == 25
and uc.get_scoring_policy(TOM)["qualify_threshold"] == 20)
edit = lc.refused(uc.edit_lead, BOSS, won.lead_id, company="Renamed")
lc.check("a converted lead is not edited", edit and edit[0] == 409, edit)
# ── 0.2.0 ───────────────────────────────────────────────────────────────────────────────
from erp_platform.repository import as_conflict
def refused(fn, *a, **kw):
try:
return lc.refused(fn, *a, **kw)
except Exception as e: # noqa: BLE001 — a lost version is a 409
if as_conflict(e) is None:
raise
return 409, as_conflict(e).reason
cy = lead_id_for("cy@bigretail.com")
v = uc.get_lead(TOM, cy)["version"]
uc.edit_lead(TOM, cy, title="Buyer", version=v)
stale = refused(uc.edit_lead, TOM, cy, title="Head buyer", version=v)
lc.check("C3 a stale version is 409 on DataK3 and writes nothing",
stale and stale[0] == 409 and uc.get_lead(TOM, cy)["title"] == "Buyer", stale)
again = uc.import_leads(MIA, "email,title\ncy@bigretail.com,Chief buyer\n", "list 3")
lc.check("C3 a re-import claims the lead's next version", again["updated"] == 1
and uc.get_lead(TOM, cy)["version"] == v + 2, (again, uc.get_lead(TOM, cy)["version"]))
bad = lc.refused(uc.import_leads, MIA, "name,email\n" + "x" * 200_000 + ",big@acme.com\n")
lc.check("H12 a row that cannot be read as CSV is 422 with its row number", bad and bad[0] == 422
and "row 2" in bad[1], bad)
from contextlib import contextmanager
eng = lc.engine(pool_size=8)
@contextmanager
def real():
with eng.begin() as conn:
yield LeadRepository(conn)
uc.repository = real
v = uc.get_lead(TOM, cy)["version"]
out = []
def race(i):
r = refused(uc.edit_lead, TOM, cy, title=f"race {i}", version=v)
out.append(r[0] if r else "ok")
ts = [threading.Thread(target=race, args=(i,)) for i in range(4)]
for t in ts:
t.start()
for t in ts:
t.join()
lc.check("C3 four edits racing from one version (real transactions): one success, the rest 409",
out.count("ok") == 1 and out.count(409) == 3, out)
return lc.finish(tables)
if __name__ == "__main__":
raise SystemExit(main())