Works / SaleQue

Client: SaleQue · AI micro-CRM · Multi-tenant RAG · AI Agents and Workflow Automation

SaleQue wanted a CRM assistant that never invents a promise. We built the guarantee into the database.

SaleQue is a micro-CRM for sales teams of 2 to 10 people. Their users kept asking one question the product could not answer: "what did we promise this client?" The answer was always in the CRM, spread across a deal, a note and an email thread. SaleQue asked Elegant to build an assistant that answers from the workspace's own records, cites every source, never sees another tenant's data, and runs at a cost a small team can pay. This is how we built it.

25tests across 8 suites, run against real PostgreSQL 17 and pgvector 0.8 in CI

0cross-tenant rows in 12 adversarial queries, with the workspace filter on and with it removed

2walls between tenants: a workspace scope in every query, Row Level Security underneath

1credit per question, 0 on a cache hit, every call priced in a ledger the customer can read

PythonFastAPIReactPostgreSQL 17pgvector 0.8Row Level SecurityClaudeOpenAIJev by TypeSafe AIPydanticpytestHybrid retrieval

The assistant inside SaleQue answering a renewal question. Retrieval is scoped to the workspace, every sentence carries a numbered source, the proposed actions become tasks in one click, and the credit ticks down once. Demo workspace data.

The client

A CRM built for teams too small to have a RevOps person.

SaleQue sells to founders and small sales teams who run their pipeline in one workspace: leads, accounts, deals, tasks and activities, with an AI layer on top that scores leads, suggests next actions and enriches records. Their roadmap had one item their in-house team had not been able to ship safely: an assistant that could be asked anything about the workspace and trusted enough to be quoted to a customer.

They had prototyped it twice. The first version chatted over the whole database and produced fluent, confident answers with no way to check them. The second summarised records into a prompt and hallucinated a discount that had never been offered. Both were pulled before launch. When SaleQue brought the brief to Elegant, the requirement had become precise.

ProductSaleQue, AI micro-CRM for teams of 2 to 10

EngagementAssistant feature, data layer to UI, fixed price, full IP transfer

StackReact frontend, Python and FastAPI backend, PostgreSQL, multi-tenant by workspace

Elegant scopeArchitecture, data model, embedding pipeline, retrieval, answer contract, AI service, lead scoring with Jev, test suite, handover

The brief

Five requirements. None of them negotiable.

Answer only from this workspace

Not from the model's general knowledge, and never from another customer's records. Isolation had to be structural, not a filter someone remembers to add.

Cite every claim

A sentence the rep cannot trace to a record must not appear. If the records do not contain the answer, the assistant says so.

Stay fresh without re-indexing everything

Reps edit records all day. Embeddings must follow content changes and ignore everything else, or the embedding bill becomes the product's biggest cost.

Cost a small team can see and control

Credits per question, a ledger per workspace, repeated questions free, and no lock-in to one model provider.

Decisions with a confidence the code can act on

Lead scores and routing must come back as typed values with a calibrated confidence, so the CRM can act when it is sure and hand the lead to a person when it is not.

What we found

A promise is never stored once.

The price is in the deal, the reason is in a note, the confirmation is in an email activity. Three record types, three screens, and the link between them existed only in the head of whoever made the promise.

CRM search is keyword search.

"Held the price" and "price freeze" never match each other. A rep who did not write the note cannot find it. Retrieval had to work on meaning and still catch exact names and amounts.

Both failed prototypes shared one flaw.

The model was trusted to decide what to say. In the version we built, the model proposes and code verifies. Anything the code cannot trace to a retrieved record is removed before render.

A score is not a sentence.

Asking a chat model to rate a lead returns a paragraph and a number with no calibration behind it. You cannot put a threshold on that. Scoring and routing needed a model built to return typed decisions, so we split the work: Jev decides, Claude and OpenAI write.

The model proposes. The code verifies. Anything that cannot be traced to a record in this workspace is removed before it reaches the screen.

Architecture

Everything lives in Postgres, on purpose.

SaleQue already ran a React frontend on a Python and FastAPI backend over PostgreSQL. Adding a separate vector database, a message broker and a cache cluster would have tripled what a 2-person team has to operate. We put the embeddings in pgvector, the queue in an outbox table, the cache, the ledger and the lead scores in ordinary tables, and enforced tenancy with Row Level Security. One database, one backup, one thing to page someone about. Model calls leave the API through one service: Claude and OpenAI for language, Jev for decisions.

One database holds records, vectors, queue, cache, ledger and lead scores. The FastAPI service is the only component allowed to talk to a model provider. Claude answers, OpenAI takes over if Claude is unavailable, Jev scores and routes. In CI, recorded HTTP payloads stand in for all three, so the suite runs offline.

Deep dive 01 · Tenant isolation

Two walls, because one filter is one bug away.

Every table that can hold customer text carries a workspace_id, and every query the application runs starts with that filter. That is the first wall, and on its own it is what most multi-tenant products rely on. We were not willing to let SaleQue's safety depend on every future query being written correctly.

The second wall is Row Level Security inside Postgres. Every FastAPI request opens one transaction, switches to the application role, sets the current workspace as a transaction-local setting, and from that point the database refuses to return or accept rows from any other workspace, even if a query forgets the filter entirely. Retrieval, full-text search, the embeddings table, the outbox, the cache and the ledger all sit behind the same policy.

sql/schema.sql · embeddings, indexes and the Row Level Security policy on every table-- one rendered document per record, embedded once per content change create table embeddings ( id bigserial primary key, workspace_id uuid not null references workspaces(id) on delete cascade, record_type text not null check (record_type in ('lead','account','deal','task','activity')), record_id uuid not null, content text not null, content_hash bytea not null, model text not null, embedding vector(1536) not null, tsv tsvector generated always as (to_tsvector('english', content)) stored, updated_at timestamptz not null default now(), unique (workspace_id, record_type, record_id) ); create index embeddings_hnsw on embeddings using hnsw (embedding vector_cosine_ops); create index embeddings_tsv on embeddings using gin (tsv); create index embeddings_ws on embeddings (workspace_id); -- second wall: the database enforces the tenant even when a query forgets the filter do $$ begin if not exists (select 1 from pg_roles where rolname = 'app_user') then create role app_user nologin; end if; end $$; grant usage on schema public to app_user; grant select, insert, update, delete on all tables in schema public to app_user; grant usage, select on all sequences in schema public to app_user; do $$ declare t text; begin foreach t in array array['leads','accounts','deals','tasks','activities','embeddings','embed_outbox','lead_scores','ai_ledger','answer_cache'] loop execute format('alter table %I enable row level security', t); execute format('alter table %I force row level security', t); execute format('create policy %I_ws on %I using (workspace_id = current_setting(''app.workspace_id'', true)::uuid) with check (workspace_id = current_setting(''app.workspace_id'', true)::uuid)', t, t); end loop; end $$;

Both settings are transaction-local, so a pooled connection can never carry one tenant's scope into the next request. The workspace itself never comes from the request. FastAPI resolves it from the signed session, and a body that tries to name a workspace is rejected with a 422 before any query runs.

app/db.py · every request runs in one transaction, as one role, for one workspace@contextmanager def as_workspace(conn: psycopg.Connection, workspace_id: str) -> Iterator[psycopg.Connection]: """Every request runs here: one transaction, the application role, one workspace. Both settings are transaction-local, so a pooled connection can never carry one tenant's scope into the next request. """ with conn.transaction(): conn.execute("set local role app_user") conn.execute("select set_config('app.workspace_id', %s, true)", [str(workspace_id)]) yield conn

app/api.py · the workspace comes from the signed session, never from the requestdef current_workspace(svc: Annotated[Services, Depends(services)], authorization: Annotated[str, Header()] = "") -> str: """The workspace comes from the verified session token. Never from the body, never from a query string.""" token = authorization.removeprefix("Bearer ").strip() ws, _, _ = token.partition(".") if not ws or not hmac.compare_digest(token, sign(ws, svc.secret)): raise HTTPException(401, "invalid session") return ws class AskIn(BaseModel): model_config = ConfigDict(extra="forbid") # a smuggled workspace_id is a 422, not a silent override question: str = Field(min_length=3, max_length=500)

The test that runs on every commit

Two seeded workspaces with overlapping company names and identical amounts. 12 adversarial questions ask workspace A about facts that only exist in workspace B, by name, by amount, by email address and by paraphrase. Then the same search with the workspace filter deliberately removed. The assertion is 0 rows from B in both cases, plus a rejected insert when a row is tagged with the wrong workspace.

Why the policy shows up in the query plan

The One-Time Filter line in the plan below is Postgres evaluating the RLS policy before the scan. It is not application code. If SaleQue's team writes a new query next year and forgets the filter, that line is still there.

The reference build's test suite and query plans, captured from a real run on PostgreSQL 17 with pgvector 0.8. 25 tests pass, the unfiltered search returns 0 rows from the other workspace, and the foreign-workspace insert is rejected by the policy.

saleque-assistant · python -m scripts.explainzsh

1. scoped retrieval, EXPLAIN ANALYZE  (3,010 embeddings, 2 workspaces, HNSW on cosine)
   Limit (actual time=14.471..14.479 rows=8 loops=1)
     ->  Sort (actual time=14.468..14.471 rows=8 loops=1)
           Sort Key: ((embedding <=> $question::vector))
           Sort Method: top-N heapsort  Memory: 17kB
           ->  Result (actual time=0.121..13.269 rows=1506 loops=1)
                 One-Time Filter: ((current_setting('app.workspace_id'::text, true))::uuid = '4f2a6c1e-8b3d-4e7a-9c21-0d5f7a3b9e11'::uuid)
                 ->  Index Scan using embeddings_ws on embeddings (actual time=0.030..1.146 rows=1506 loops=1)
                       Index Cond: (workspace_id = '4f2a6c1e-8b3d-4e7a-9c21-0d5f7a3b9e11'::uuid)
   Planning Time: 0.424 ms
   Execution Time: 14.730 ms
   note: the One-Time Filter line is the RLS policy; for a 1.5k-row tenant the planner picks the workspace index and an exact top-N sort

1b. same search across the whole table  (large tenant or admin path: the planner switches to the HNSW index)
   Limit (actual time=1.045..1.145 rows=8 loops=1)
     ->  Index Scan using embeddings_hnsw on embeddings (actual time=1.042..1.138 rows=8 loops=1)
           Order By: (embedding <=> $question::vector)
   Planning Time: 0.088 ms
   Execution Time: 1.180 ms

2. second wall  (same table, NO workspace filter, running as app_user for workspace A)
   rows returned: 50   from workspace B: 0   policy: workspace_id = current_setting('app.workspace_id')

3. same query as the database owner  (what the policy protects against)
   4f2a6c1e  1506 embeddings
   b7e1d2c3  1504 embeddings

4. write with a foreign workspace_id, as app_user for A
   rejected  new row violates row-level security policy for table "tasks"

Deep dive 02 · Embedding lifecycle

Re-embed when the words change. Nothing else.

Reps touch records constantly: reassign an owner, move a stage, tick a task. Almost none of that changes what the record says. Each record type has a template that renders it into one readable paragraph, the same text the assistant will later cite. We hash that text with SHA-256 and store the hash beside the vector. On every write, the new hash is compared with the stored one. Same hash, no work. Different hash, one row goes into the outbox inside the same transaction as the record write, so there is no window where a record exists without its embedding job.

A Python worker claims pending rows with FOR UPDATE SKIP LOCKED, so several workers can run without ever taking the same job, calls the OpenAI embeddings adapter and upserts into the embeddings table with the workspace id, the record type and id, the content, the hash and the model version. The model version column means a future embedding model upgrade is a progressive backfill, not a big-bang re-index. Deleting a record fires a trigger that removes its embedding in the same transaction, so a removed note can never be retrieved.

app/outbox.py · enqueue only when the rendered document changed, drain with SKIP LOCKEDdef enqueue_if_changed(conn: Connection, ws: str, kind: RecordType, record_id: str) -> bool: """Called inside the write transaction. Enqueues only when the rendered document changed.""" row = _row(conn, kind, record_id) if row is None: return False digest = hashlib.sha256(render_document(kind, row).encode()).digest() prev = conn.execute( "select content_hash from embeddings where record_type = %s and record_id = %s", [kind, record_id] ).fetchone() if prev and bytes(prev["content_hash"]) == digest: return False conn.execute( "insert into embed_outbox (workspace_id, record_type, record_id) values (%s, %s, %s)", [ws, kind, record_id] ) return True def drain_outbox(conn: Connection, ws: str, embedder: Embedder, batch: int = 100) -> int: """Worker: claim pending rows (safe with many workers), embed, upsert, mark done.""" pending = conn.execute( """select id, record_type, record_id from embed_outbox where processed_at is null order by id limit %s for update skip locked""", [batch], ).fetchall() for p in pending: row = _row(conn, p["record_type"], p["record_id"]) if row is not None: doc = render_document(p["record_type"], row) conn.execute( """insert into embeddings (workspace_id, record_type, record_id, content, content_hash, model, embedding) values (%s, %s, %s, %s, %s, %s, %s::vector) on conflict (workspace_id, record_type, record_id) do update set content = excluded.content, content_hash = excluded.content_hash, model = excluded.model, embedding = excluded.embedding, updated_at = now()""", [ws, p["record_type"], p["record_id"], doc, hashlib.sha256(doc.encode()).digest(), embedder.model, to_vec(embedder.embed(doc))], ) conn.execute("update embed_outbox set processed_at = now() where id = %s", [p["id"]]) return len(pending)

Verified in the suite

An update to updated_at enqueues nothing. An update to notes enqueues exactly once, the worker drains exactly one row, and the stored hash changes. A deleted lead leaves no embedding behind. Three tests, tests/test_lifecycle.py.

Why an outbox and not a queue service

The outbox row and the record write commit together or not at all. There is no second system to keep in sync, no retry topology to design, and the worker is a loop over one table. For a 2-person team this is the difference between a feature they can operate and one they cannot.

Deep dive 03 · Retrieval

Meaning and exact match, fused.

A vector search finds "held the price" when the rep asks about a "price freeze". It is bad at "$48,000" and at a person's name. Postgres full-text search is the reverse. So every question runs both in one SQL round trip, each scoped to the workspace, and the two lists are merged with reciprocal rank fusion inside Postgres. The top 8 records go to the model and nothing else does.

The plans tell an honest story about scale. For a tenant with 1,500 embeddings the planner chose the workspace index and an exact top-N sort, about 15 ms with 100% recall. Across the whole table it switched to the HNSW index at about 1 ms. Small tenants get exact answers, large tenants get the approximate index, and nobody had to tune anything.

app/retrieve.py · vector and full text search, both scoped, fused by reciprocal rank in one query# Vector search and full text search in one round trip, fused with reciprocal rank fusion (k = 60). # The workspace filter is the first wall. RLS underneath makes it redundant on purpose. HYBRID = """ with q as (select %(vec)s::vector as v, plainto_tsquery('english', %(text)s) as t), vec as ( select record_type, record_id, content, row_number() over (order by embedding <=> q.v) as r from embeddings, q where workspace_id = %(ws)s order by embedding <=> q.v limit 16), fts as ( select record_type, record_id, content, row_number() over (order by ts_rank(tsv, q.t) desc) as r from embeddings, q where workspace_id = %(ws)s and tsv @@ q.t order by ts_rank(tsv, q.t) desc limit 16) select record_type, record_id::text, content, round(sum(1.0 / (60 + r))::numeric, 4)::float as score from (select * from vec union all select * from fts) both_lists group by record_type, record_id, content order by score desc, record_id limit %(k)s """ def retrieve(conn: Connection, ws: str, question: str, embedder: Embedder, k: int = 8) -> list[Hit]: rows = conn.execute(HYBRID, {"vec": to_vec(embedder.embed(question)), "text": question, "ws": ws, "k": k}).fetchall() return [Hit(**r) for r in rows]

Deep dive 04 · The answer contract

The model fills in a form. The code marks it.

The prompt contains the question and the retrieved records, and nothing else. The model must return a fixed shape: an array of sentences, each with the ids of the records it relied on, and up to 3 proposed actions, each typed as a task, a deal update or a calendar hold. Free text outside the schema is rejected. The schema is one Pydantic model, and that same model generates the JSON schema sent to the provider, so the contract and the validator can never drift apart.

Then the verifier runs. Every cited id is checked against the set that was actually retrieved for this question. A sentence with no valid source is dropped. A sentence citing a record that was never retrieved is dropped. Citations are renumbered in order of first use so the UI can show [1], [2], [3]. If nothing survives, the response is the not-found state, with the closest records the search did find, so the rep can judge whether the promise was ever written down. The same verify-before-render rule now runs the email and meeting pipeline we built for OfficeSyncPro.

saleque-assistant · python -m scripts.askzsh

ask  ws 4f2a6c1e · 6 records retrieved in 2 ms · answer miss in 3 ms
q   What did we promise Northwind before the renewal call?

  Three commitments are on record for Northwind. [1][2][3]
  Renewal price held at $48,000 for 12 months, confirmed by email on 9 Sep. [1]
  Two extra seats included at no charge, agreed on the 12 Sep call with Maya Chen. [2]
  An updated SOW was promised by 20 Sep. That task is still open. [3]
  The deal sits in Negotiation with a close date of 30 Sep. [4]

  dropped  Northwind also asked about an API add-on.  (no source in retrieved set)
  dropped  They mentioned a competitor quote.  (no source in retrieved set)

sources
  [1] activity 9b2d5f6a  Email on 2026-09-09: Pricing confirmation to Maya Chen. Confirmed the renewal …
  [2] activity e1d2c3b4  Note on 2026-09-12: Call notes: renewal scope. Call with Maya Chen. Agreed two…
  [3] task     c56a4180  Task Send updated SOW to Maya Chen, open, due 2026-09-20, owner Sam.
  [4] deal     3f2504e0  Deal Northwind renewal, $48,000, stage Negotiation, closes 2026-09-30. Annual …

proposed actions
  • task          Send the updated SOW to Maya Chen  due today
  • deal_update   Add 2 seats to the Northwind renewal deal
  • calendar_hold Book the renewal call before 30 Sep

ledger  feature=assistant.ask provider=test cache=miss credits=1  balance=199

The python -m scripts.ask command in the reference build. Two of the seven sentences the completer returned had no source in the retrieved set. Both were dropped before render. The credit ticked once and the answer was cached against the question and the record ids.

app/answer.py · the answer schema and the verifierclass Answer(BaseModel): """The only shape a model is allowed to return. Its JSON schema is sent to the provider.""" model_config = ConfigDict(extra="forbid") sentences: list[Sentence] actions: list[Action] = Field(default_factory=list, max_length=3) def verify(raw: object, retrieved: list[Hit]) -> Verified: """Every cited id must be in the retrieved set. A sentence with no valid source is dropped. Citation numbers follow first use, so the UI renders [1] [2] in reading order. """ parsed = Answer.model_validate(raw) allowed = {h.record_id: h for h in retrieved} order: list[str] = [] def num(record_id: str) -> int: if record_id not in order: order.append(record_id) return order.index(record_id) + 1 kept, dropped = [], [] for s in parsed.sentences: ok = [i for i in s.sources if i in allowed] if ok: kept.append(CitedSentence(text=s.text, sources=[num(i) for i in ok])) else: dropped.append(s.text) actions = [a for a in parsed.actions if a.sources and all(i in allowed for i in a.sources)] return Verified(sentences=kept, dropped=dropped, actions=actions, citations=[allowed[i] for i in order], not_found=not kept)

Deep dive 05 · The AI service

One door for every model call.

No feature in SaleQue calls a provider directly. Everything goes through one service that does 4 things on every call. It meters the call against the workspace at a fixed credit price the customer can see. It checks a cache keyed on the workspace, the normalised question, the sorted ids of the retrieved records and the model, so the same question over the same evidence is free and instant, and a changed record misses on purpose. It routes through one adapter interface. Claude is the primary and is forced into the answer schema with a tool call. If Claude returns an outage, a rate limit or a timeout, the same request goes to OpenAI with a JSON schema response format, and the same verifier runs on whatever comes back. The ledger records which provider actually answered. And it refuses cleanly: a workspace at 0 credits sees a clear state instead of a failed request, while embedding of changed records continues so retrieval quality does not decay while the customer decides whether to top up.

app/ai_service.py · meter, cache, route, refusedef ask(conn: Connection, ws: str, question: str, retrieved: list[Hit], completer: Completer) -> AskResult: """One door for every model call: meter, cache, route, refuse.""" evidence = ",".join(sorted(h.record_id for h in retrieved)) key = hashlib.sha256("|".join([ws, normalise(question), evidence, completer.model]).encode()).hexdigest() cached = conn.execute("select answer from answer_cache where key = %s and expires_at > now()", [key]).fetchone() if cached: ledger(conn, ws, "assistant.ask", completer.provider, completer.model, 0, 0, "hit", 0) return AskResult(**cached["answer"], cache="hit", credits=0) if credits(conn, ws) < PRICE["assistant.ask"]: raise OutOfCredits(ws) out = completer.complete(question, retrieved) verified = verify(out.raw, retrieved) conn.execute( "insert into answer_cache (key, workspace_id, answer, expires_at) values (%s, %s, %s, now() + interval '24 hours')", [key, ws, Jsonb(verified.model_dump())], ) ledger(conn, ws, "assistant.ask", out.provider, out.model, out.input_tokens, out.output_tokens, "miss", PRICE["assistant.ask"]) return AskResult(**verified.model_dump(), cache="miss", credits=PRICE["assistant.ask"])

app/providers.py · Claude held to the schema, OpenAI as the fallbackclass ClaudeCompleter: """Claude with a forced tool call, so the reply is schema-shaped JSON, never prose.""" provider = "anthropic" def __init__(self, client: anthropic.Anthropic, model: str = "claude-sonnet-5") -> None: self.client, self.model = client, model def complete(self, question: str, records: list[Hit]) -> Completion: msg = self.client.messages.create( model=self.model, max_tokens=1024, system=SYSTEM, tools=[{"name": "answer", "description": "Return the cited answer.", "input_schema": ANSWER_SCHEMA}], tool_choice={"type": "tool", "name": "answer"}, messages=[{"role": "user", "content": user_prompt(question, records)}], ) raw = next(b.input for b in msg.content if b.type == "tool_use") return Completion(raw, self.provider, msg.model, msg.usage.input_tokens, msg.usage.output_tokens) class WithFallback: """Primary first. On a provider outage, rate limit or timeout, the fallback answers instead.""" def __init__(self, primary: Completer, fallback: Completer) -> None: self.primary, self.fallback = primary, fallback self.provider, self.model = primary.provider, primary.model def complete(self, question: str, records: list[Hit]) -> Completion: try: return self.primary.complete(question, records) except (anthropic.APIStatusError, anthropic.APIConnectionError, openai.APIStatusError, openai.APIConnectionError): return self.fallback.complete(question, records)

Deep dive 06 · Decisions with Jev

Language models write. Jev decides.

Every new lead in SaleQue needs three decisions: how well it fits the workspace's ideal customer, what the sender wants right now, and whether money is mentioned. We did not want a chat model writing a paragraph and a number for that. The API sends the lead and the workspace's ideal customer profile to Jev, TypeSafe AI's System One model, with three typed questions: a score on a 4-level rubric, a choice between 4 intents, and a yes or no on budget. Jev returns each answer as a typed value with a confidence, through the official Python SDK.

The routing rule is plain Python that anyone on SaleQue's team can read. Below 0.80 confidence on intent, the lead goes to a person and nothing is guessed. Confident noise is archived. Support requests go to customer success. A buying lead with a good fit goes to sales at high priority. Every decision is stored in lead_scores with the raw answers and the model version, behind the same RLS policy as everything else, and metered in the same ledger as the language models.

app/jev.py · typed questions to Jev, and the routing ruleLEAD_QUESTIONS = { "fit": Score( instructions="How well does this lead match the workspace's ideal customer profile?", criteria=["No match", "Weak match", "Good match", "Strong match"], ), "intent": Choice( instructions="What does the lead want right now?", criteria={ "buying": "Asks for pricing, a proposal or a start date", "evaluating": "Comparing options, no timeline yet", "support": "Existing customer with a problem", "noise": "Spam, recruiting or unrelated", }, ), "has_budget": Noul(instructions="The lead mentions a budget, funding or an approved spend"), } AUTO = 0.80 # below this confidence, a person decides @dataclass(frozen=True) class Route: owner: str # sales | success | review | archived priority: str # high | normal def decide(res: SystemOneResponse) -> Route: intent, fit, budget = res.choices["intent"], res.scores["fit"], res.nouls["has_budget"] if intent.confidence < AUTO: return Route("review", "normal") if intent.choice == "noise": return Route("archived", "normal") if intent.choice == "support": return Route("success", "normal") hot = fit.score >= 2.0 and (intent.choice == "buying" or budget.noul >= 0.7) return Route("sales", "high" if hot else "normal")

saleque-assistant · python -m scripts.score_leadszsh

score-leads  model=jev-1.13  auto>=0.8  ws 4f2a6c1e

  16fd2706  Daniel Reyes     buying      0.93  fit 2.61  budget 0.91  -> sales/high
  5e8b1c2d  Priya Nair       evaluating  0.58  fit 1.12  budget 0.04  -> review
  0c1d2e3f  SEO Growth Team  noise       0.97  fit 0.06  budget 0.00  -> archived

ledger  feature=lead.score provider=typesafe calls=3 tokens=750  stored in lead_scores, scoped by RLS

The python -m scripts.score_leads command in the reference build, run on recorded System One responses. The vague lead came back at 0.58 confidence on intent, so it went to review instead of being guessed. The spam was archived. The funded founder went to sales at high priority.

Why not ask Claude for the score

A chat model can produce a number, but not a calibrated one, and the same lead can come back with a different number on a retry. Jev's confidence is what makes the 0.80 gate mean something. It also keeps the language models on the work they are good at: reading records and writing cited answers.

Verified in the suite

The typed questions reach /v1/systemone. A hot lead routes to sales. A 0.58 confidence routes to review. Noise is archived. Every decision is stored and metered inside the workspace. Four tests, tests/test_jev.py.

Product surfaces

What SaleQue's users see.

The assistant uses the same components as the rest of the CRM, so an AI answer never looks like a widget bolted on. Six surfaces carry the whole feature.

Surface 01

One box, one workspace.

The ask box sits on the home screen next to the day's tasks. Prompt templates cover the questions reps ask every week. The scope line under the box says exactly which workspace the assistant can see and what a question costs. There is no "search everything" mode.

Surface 02

Cited, or not shown.

Each sentence carries its numbered sources. Clicking a citation opens the record in a side panel. The struck-through line is what the verifier removed: a sentence the model wrote with no record behind it. In production it is never rendered. It is shown here because it is the whole point.

assistant · verified answer

Three commitments are on record for Northwind.123
Renewal price held at $48,000 for 12 months, confirmed by email on 9 Sep.1
Two extra seats included at no charge, agreed on the 12 Sep call with Maya Chen.2
An updated SOW was promised by 20 Sep. That task is still open.3
Northwind also asked about an API add-on. dropped · no source

1 · Email

Pricing confirmation to Maya Chen

9 Sep 2026 · activity

2 · Note

Call notes · renewal scope

12 Sep 2026 · activity

3 · Task

Send updated SOW to Maya Chen

Due 20 Sep · open

4 · Deal

Northwind renewal · $48,000

Negotiation · closes 30 Sep

Surface 03

Actions are typed, proposed, and confirmed by a person.

The same structured response carries up to 3 actions. Each has a type the CRM already understands. Nothing is written automatically. The rep clicks once, and the created task links back to the records that justified it, which makes the CRM richer every time the assistant is used.

Surface 04

The not-found state is a real screen.

When nothing survives verification, the assistant says so and shows the closest records it did retrieve, cited, so the rep can decide whether the promise was ever written down. This state has its own design because in a CRM it happens often, and it is where the second prototype lied.

Surface 05

A ledger the customer can read.

Every model call is a row: feature, provider, tokens, cache result, credits. Workspace admins see spend per feature and per day. Support sees the same table when a customer asks why a credit was used. The number in the top bar is this table, summed.

settings · billing · ai ledger199 credits

time	feature	provider	tokens in / out	cache	credits
21:44:08	assistant.ask	anthropic	3,912 / 418	miss	1
21:44:31	assistant.ask	anthropic	0 / 0	hit · 24h	0
21:39:17	embed.record	openai	412 / 0	miss	0
21:12:02	lead.score	typesafe	212 / 38	bypass	0

Surface 06

A lead inbox that knows when it is unsure.

Every inbound lead arrives scored and routed. The confidence sits next to the decision, so a rep can see why a lead jumped the queue. Anything under 0.80 lands in a review lane with the typed answers attached, and the rep makes the call.

leads · inbox · scored by jev

lead	intent	conf	fit	route
Daniel Reyes · Tallyroom	buying	0.93	2.61	sales · high
Priya Nair · Oakline	evaluating	0.58	1.12	review
SEO Growth Team	noise	0.97	0.06	archived

jev-1.13 · auto route at 0.80 confidence and above

Verified properties

What the test suite proves, file by file.

These are the guarantees SaleQue's team can rely on, and the tests that enforce them on every commit. The suite runs the real schema on PostgreSQL 17 with pgvector 0.8. Claude, OpenAI and Jev run through their real SDKs against recorded HTTP payloads, and a deterministic embedder stands in for the embeddings API, so CI runs offline and the results are reproducible.

Property	How it is checked	Suite
The workspace comes from the session only	A body naming another workspace returns 422. A forged token returns 401. Scoring another workspace's lead returns 404.	`tests/test_api.py`
Workspace A can never retrieve a workspace B record	12 adversarial questions by name, amount, email and paraphrase. 0 foreign rows.	`tests/test_isolation.py`
Forgetting the filter does not leak	The same search with no WHERE clause, as the app role. 0 foreign rows.	`tests/test_isolation.py`
A row tagged with another workspace is rejected	Insert as workspace A with workspace B's id. Policy violation.	`tests/test_isolation.py`
Metadata edits do not re-embed	Touching updated_at enqueues nothing. Editing notes enqueues once and changes the hash.	`tests/test_lifecycle.py`
Deleted records disappear from retrieval	Delete a lead. The trigger removes its embedding in the same transaction.	`tests/test_lifecycle.py`
The renewal question brings the 4 relevant records	Email, note, task and deal all in the top 8. Exact amount found by full text.	`tests/test_retrieval.py`
Uncited sentences never render	7 sentences in, 5 out, the 2 without a valid source dropped. Not-found when 0 survive. Off-schema output rejected.	`tests/test_contract.py`
Repeat questions cost nothing, changed evidence costs 1	Miss then hit on the same question, balance 199. Different record set misses again. A workspace at 0 credits is refused before any model is called.	`tests/test_ledger.py`
Claude is held to the schema, OpenAI takes over on an outage	Real Anthropic and OpenAI SDKs on recorded payloads. The forced tool call is checked. A 529 from Claude falls back to OpenAI under the same contract.	`tests/test_providers.py`
Uncertain decisions go to a person	Typed questions reach Jev. Intent at 0.58 routes to review, noise is archived, a hot lead goes to sales, every decision is stored and metered.	`tests/test_jev.py`

What the numbers on this page are. The test counts, query plans and terminal output were captured from the reference build described above. The 2 ms retrieval and the 15 ms and 1 ms plan timings are from that run on seeded data. Provider and Jev responses in the tests are recorded payloads, not live calls. They show the shape of the system, not SaleQue's production traffic, which is theirs to publish.

What we deliberately did not build

Four things a demo would have included.

No fine-tuning

The value sits in the retrieval and the answer contract. A tuned model would have added cost without adding trust, tied SaleQue to one vendor, and made the provider swap meaningless.

No cross-workspace intelligence

"Deals like this usually close at" would be a good feature. It also requires reading other tenants' data. It stays out until it can be built on aggregates that never touch a record.

No answers from general knowledge

The model is not allowed to fill gaps. A question the workspace cannot answer gets the not-found state with the closest cited records.

No autonomous writes

Actions are proposed and typed. A person confirms every one. The CRM stays the system of record and the assistant does not get to edit it on its own.

Delivered to SaleQue

A feature their team can run without us.

Schema and migrationsTables, RLS policies, indexes and the app role, as versioned migrations in their repo

Embedding pipelineRender templates per record type, hashing, outbox and worker, backfill command

Retrieval and answer contractHybrid search, fusion, schema, verifier, not-found state

AI serviceLedger, cache, Claude and OpenAI adapters with fallback, Jev lead scoring, credit pricing per feature, admin views

Test suite in CI25 tests on real Postgres, offline, deterministic, runs on every pull request

Runbook and decision recordsHow to rotate a provider, re-embed on a model change, read the ledger, add a record type

How the build ran

Six phases, one senior team, weekly demos to SaleQue.

Phase 01

Data model and isolation first

workspace_id on every table, RLS policies, the app role and the adversarial isolation suite, all before a single embedding existed. The most important tests were green before the feature had a UI.

Phase 02

Embedding lifecycle

Record templates, hashing, outbox and worker, model version column, backfill of SaleQue's existing workspaces.

Phase 03

Retrieval and the contract

Hybrid search and fusion, the response schema, the verifier and the not-found state, evaluated against a hand-written question set over seeded workspaces.

Phase 04

Assistant surfaces

Ask box, templates, answer rendering with citations, source panel, proposed actions to tasks, built with SaleQue's existing components.

Phase 05

AI service and decisions

Ledger, cache, Claude and OpenAI adapters with fallback, Jev lead scoring with the 0.80 review gate, per-feature pricing, the 0-credit state, admin spend view.

Phase 06

Hardening and handover

Outbox under bulk import, rate limits per workspace, audit log of every assistant call, runbook, decision records, walkthrough with SaleQue's engineers.

Tech stack

PostgreSQL 17 + pgvector 0.8Records, vectors, outbox, cache, ledger and lead scores in one database. RLS on every table.

Python and FastAPIThe API and the AI service. Typed routes, psycopg 3, the workspace resolved from the signed session.

ReactSaleQue's existing frontend. The assistant and the lead inbox reuse its components.

Claude and OpenAIClaude answers through a forced tool call. OpenAI is the fallback and the embeddings provider.

Jev by TypeSafe AILead fit, intent and budget as typed answers with calibrated confidence, through the Python SDK.

Pydantic and pytestThe answer contract and its JSON schema in one model. 25 tests on a real Postgres, offline in CI.

Takeaway

RAG is a data problem wearing an AI costume.

Almost none of the work on this project was prompt engineering. It was deciding what a record looks like as text, when it changes, who is allowed to see it, and what the model is allowed to return. Get those 4 things right and the assistant is boring in the best way: it answers, it shows its sources, and it says "I could not find that" instead of guessing.

If your product already holds the data your customers keep asking about, this is a 6-week feature with a fixed price. The tests on this page are the ones we would write for yours.

The service behind this build

AI Agents and Workflow Automation

Retrieval, agents and typed AI features built into your product by one senior team. Fixed price, fixed timeline, full IP transfer.

Explore service ↗

Adding this to a product that is already live? See AI Integration for Existing Products

Next case studyOfficeSyncPro: every email and meeting, turned into work you approveAI on Microsoft Graph for a live Microsoft 365 product: triage, commitment extraction with receipts, Teams summaries, follow-ups and scheduling.Read the case study ↗

OfficeSyncPro AI suggestions panel beside a client email

Ready to build something like this?

Book a free 15-minute call. We will look at what you are building and tell you honestly what it takes and whether we are the right fit.

Book a free call ↗See our services