2026-09-17, Engineering · Rank Tracking

Programmatic AI visibility tracking: system design

Programmatic AI visibility tracking means collecting AI answers for a large, versioned set of prompts on a schedule, across engines and markets, and turning them into metrics without a person in the loop. At a few dozen prompts a cron job and a spreadsheet work. At thousands of prompts, several clients and daily runs, you need a system: planning, submission, delivery, reconciliation, storage and reporting, each with a clear contract.

The design below assumes you have decided what to measure (AI rank tracking and the share of voice framework cover metrics). Code is Python with requests against the HTTP API.

The architecture

The dashed path is the recovery route: the plan table records what should exist, and the reconciler fetches whatever the webhook never delivered.

Every arrow is a durable handoff: a row in a table or an object in storage. Nothing depends on a process staying alive between submission and delivery, which matters because a task may wait in the queue and a webhook may be retried for tens of minutes.

1. The prompt catalog

The catalog is the definition of what you measure, so it lives in version control, not in a dashboard form.

A catalog record:

- id: p0412
  text: "best payroll software for a restaurant with 30 employees"
  cluster: payroll-smb-category
  intent: category
  client: acme
  brands: [acme-payroll, globex, initech]
  engines: [CHATGPT, GEMINI]
  markets: [{country: US}]
  runs_per_day: 2
  added: 2026-09-01

Rules:

• Ids are permanent. Rewording a prompt creates a new id; retire the old one. Otherwise history silently changes meaning.
• Brand sets are versioned with the prompt. Mention rank depends on who you track.
• Markets are explicit. country is required on the assistant endpoints: ChatGPT, Gemini and AI Mode answer from 213 countries, and Copilot, Perplexity and Grok take any country; Gemini also takes a US state. Market is part of the task identity, not a filter applied later.

Prompt set design covers sourcing and clustering.

2. The planner and the plan table

The planner expands the catalog into concrete tasks for a run window and writes them to a plan table before anything is submitted. The plan table is what makes the rest of the system checkable: at any moment you can compare what should exist with what arrived.

create table plan (
  idempotency_key text primary key,        -- p0412-chatgpt-US-2026-09-17-r1
  prompt_id       text not null,
  client          text not null,
  engine          text not null,           -- CHATGPT, GEMINI, ...
  market          jsonb not null,
  run_date        date not null,
  run_no          int  not null,
  payload         jsonb not null,
  task_id         uuid,                    -- set after submission
  submit_error    text,
  status          text not null default 'PLANNED',  -- PLANNED, SUBMITTED, COMPLETED, FAILED, REFUSED
  credits_charged numeric,
  updated_at      timestamptz not null default now()
);
create index on plan (status, run_date);

The idempotency key encodes meaning: prompt, engine, market, date, run. An idempotencyKey is unique across your account, so include the client id or prompt id namespace when several clients share one account. Because the key is deterministic, the planner can run twice for the same day without creating duplicate work.

3. The submitter

The submitter reads PLANNED rows, groups them into batches of up to 500, and posts them.

import os
import time
import requests

API = "https://api.answerline.dev"
HEADERS = {"Authorization": f"Bearer {os.environ['ANSWERLINE_API_KEY']}"}
WEBHOOK = "https://hooks.example.com/answers"


def submit(rows):
    """rows: plan rows with idempotency_key, engine, payload. Returns per-row outcomes."""
    body = [{
        "taskType": r["engine"],
        "payload": r["payload"],
        "idempotencyKey": r["idempotency_key"],
        "webhook": {"url": WEBHOOK},
    } for r in rows]

    while True:
        resp = requests.post(f"{API}/v1/async/task/batch", headers=HEADERS, json=body, timeout=60)
        if resp.status_code == 429:
            err = resp.json()["error"]
            if err["code"] == "QUEUE_LIMIT_EXCEEDED":
                return {"queue_full": err.get("details", {}).get("remainingCapacity")}
            time.sleep(int(resp.headers.get("Retry-After", "1")))
            continue
        resp.raise_for_status()
        break

    outcomes = []
    for item in resp.json()["results"]:
        row = rows[item["index"]]
        if item["success"]:
            outcomes.append((row["idempotency_key"], "SUBMITTED", item["task"]["id"], None))
        elif item["error"]["code"] == "RESOURCE_ALREADY_EXISTS":
            outcomes.append((row["idempotency_key"], "SUBMITTED", None, None))  # created by an earlier attempt
        else:
            outcomes.append((row["idempotency_key"], "REFUSED", None, item["error"]["code"]))
    return {"outcomes": outcomes}

Handle each response class deliberately:

A network timeout on the batch request itself is safe to retry with the same body: items that were created come back as RESOURCE_ALREADY_EXISTS, and nothing runs twice. That is the payoff of deterministic keys. A request whose every task is refused claims no keys, so the same keys can be sent again after topping up the balance (async docs).

Use priority (1 to 10, higher first) to let an ad-hoc client request jump ahead of the nightly bulk run in your own queue; it does not affect other accounts.

4. Capacity planning

There is no per-account request rate limit and no concurrency cap. Two limits bound the system (rate limits):

• Queue capacity: bounded only by the memory queued and running tasks take, about 100,000 tasks of a typical prompt, more of shorter ones. GET /v1/async/status reports queuedTasks, processingTasks and a breakdown by priority.
• Task deadlines: a task has 4 hours from creation to start and five minutes from first start to finish; a task that never started in time fails uncharged with QUEUE_WAIT_EXCEEDED.

Measure how long a full run takes from first submission to last delivery, and per-task durations from task.latencyMs on your own completed tasks, per engine; don’t assume a number. Then check the daily plan fits your collection window. If it doesn’t, spread runs across the day (which also samples more answer variance) or reduce runs per prompt before adding markets. Sampling vs census explains why more prompts often beat more runs.

Throttle submission to the queue rather than dumping a week of work at once. A simple rule: submit the next batch when queuedTasks falls below a few batches’ worth. Tasks queued far ahead of time risk the 4-hour start bound.

5. The webhook receiver

Each finished task is posted to the task’s webhook.url with the same JSON that GET /v1/async/task/{id} returns: task, credits and response (webhooks). The receiver’s only job is to verify, store and acknowledge quickly.

import hashlib
import hmac
import json
import os
import time
from flask import Flask, abort, request

app = Flask(__name__)
SECRET = os.environ["WEBHOOK_SECRET"].encode()


def verified(raw: bytes, header: str | None, tolerance: int = 300) -> bool:
    if not header:
        return False
    parts = [p.split("=", 1) for p in header.split(",") if "=" in p]
    ts = [v for k, v in parts if k == "t"]
    sigs = [v for k, v in parts if k == "v1"]
    if len(ts) != 1 or not sigs or abs(time.time() - int(ts[0])) > tolerance:
        return False
    expected = hmac.new(SECRET, ts[0].encode() + b"." + raw, hashlib.sha256).hexdigest()
    return any(hmac.compare_digest(expected, s) for s in sigs)


@app.post("/answers")
def answers():
    raw = request.get_data()
    if not verified(raw, request.headers.get("Webhook-Signature")):
        abort(400)
    delivery = json.loads(raw)
    if delivery.get("test"):
        return "", 204
    store_raw(delivery["task"]["id"], raw)   # idempotent write keyed by task.id
    return "", 204

Contract points that shape the receiver:

• Verify the raw bytes before parsing; the signature is an HMAC-SHA256 of <t>.<raw body>. During a secret rotation the header can carry two v1 values; accept either.
• Answer 2xx within 15 seconds. Anything else is retried, up to 10 attempts over roughly 21 to 43 minutes. Do the extraction elsewhere.
• Deduplicate by task.id, which is inside the signed body. Deliveries can arrive more than once and in any order.
• Ignore test deliveries, which carry "test": true.

Webhook receiver design covers the inbox pattern in depth.

6. The reconciler

Webhooks are reliable, not guaranteed: your endpoint can be down past the retry window, or a deploy can drop requests. The reconciler closes the gap using the plan table.

1. Select SUBMITTED rows whose task_id has no raw object and whose submission is older than a threshold (for example one hour, or longer when your queue is deep).
2. GET /v1/async/task/{task_id} for each.
3. If task.status is COMPLETED or FAILED, store the body as a webhook would have.
4. If QUEUED or PROCESSING, leave it. Treat only COMPLETED and FAILED as final; a task can go from PROCESSING back to QUEUED.
5. For RESOURCE_ALREADY_EXISTS rows with no stored task_id, an earlier attempt created the task but its response was lost. Its delivery carries task.idempotencyKey, so have the receiver match deliveries to plan rows by key and fill in task_id from there.

A FAILED task is not charged. Resubmit those that failed with QUEUE_WAIT_EXCEEDED under a new run number if the window still matters; record the rest as missing observations, never as “not mentioned”.

7. Raw storage

Write the delivery body unchanged to object storage, keyed so it can be listed by the dimensions you query:

raw/engine=CHATGPT/date=2026-09-17/client=acme/<task_id>.json.gz

Raw results are the system’s source of truth. When you add an alias for a brand, change the rank definition, or start tracking a new competitor, you re-run extraction over raw history instead of losing it. Keep raw objects for as long as you want to be able to restate metrics.

8. Extraction into observations

The extractor reads new raw objects and writes narrow, typed rows. Separate tables by grain:

create table answers (
  task_id uuid primary key, idempotency_key text, prompt_id text, client text,
  engine text, market jsonb, run_date date, status text, credits_charged numeric,
  source_count int, text_length int
);
create table mentions (
  task_id uuid, brand text, mention_rank int,      -- null when absent
  primary key (task_id, brand)
);
create table citations (
  task_id uuid, position int, url text, domain text, inline boolean,
  primary key (task_id, position, url)
);
create table fanout_queries (
  task_id uuid, query text, primary key (task_id, query)
);

Field mapping per engine, from the API contract:

ChatGPT’s entities[] lists the brands and products an answer names; on the other engines, extract brand mentions from text. ChatGPT’s searchQueries[] comes back empty on the answers served today.

Write an answers row for every planned task, including failed ones with null metrics. Coverage (completed / planned) is then a query, and a collection outage never looks like a visibility drop.

9. Metrics and dashboards

Compute metrics as SQL views over observations, per client, prompt cluster, engine, market and week (prompts is the catalog loaded into a table):

create view weekly_presence as
select a.client, p.cluster, a.engine, a.market->>'country' as country,
       date_trunc('week', a.run_date) as week, m.brand,
       count(*) filter (where a.status = 'COMPLETED')                          as completed,
       count(*) filter (where m.mention_rank is not null)                      as present,
       count(*) filter (where m.mention_rank <= 3)                             as top3,
       avg(case when m.mention_rank is null then 0 else 1.0 / m.mention_rank end) as mrr
from answers a
join prompts p using (prompt_id)
join mentions m using (task_id)
group by 1, 2, 3, 4, 5, 6;

Put intervals in the presentation layer (Wilson for rates, bootstrap by prompt for MRR), and show completed beside every rate so readers see the sample size.

Dashboards that hold up:

1. Coverage panel: planned, completed, failed per engine per day. Read it first.
2. Presence and top-3 rate over time, per engine, with interval bands.
3. Competitor table: brands by presence rate and MRR for a cluster, with sample size.
4. Cited domains: top domains in citations for prompts where the client is absent.
5. Fan-out coverage: share of fanout_queries for which the client ranks in Google, from a separate GOOGLE task stream ($0.00103 per query for one page). ChatGPT contributes no rows to fanout_queries today, because its list comes back empty.
6. Accuracy: answers naming the client that contain a wrong key fact.

10. Cost control

Price per request at the Starter tier: ChatGPT $0.0006, with every include flag free; Gemini $0.0006; Copilot $0.00274; Perplexity $0.0015; Grok $0.00218; AI Mode $0.00206; Google Search $0.00103 for the first page, $0.00071 per extra page and $0.00071 for the AI Overview; Google News $0.0002, extra pages free.

Monthly cost for one catalog line:

cost = markets × engines' task cost summed × runs per day × days

Worked example: 400 prompts, 1 market (US), ChatGPT ($0.0006) and Gemini ($0.0006), 1 run per day, 30 days: 400 × 1 × $0.0012 × 30 = $14.40 a month at Starter prices. Halving runs to every other day halves it. Reserve a separate budget for fan-out checks on Google.

Guards worth building:

• A pre-flight in the planner that sums expected cost per client and refuses a plan over budget.
• GET /v1/credits before each submission cycle; remaining is the balance in dollars; stop when it minus the next cycle’s planned cost goes negative, or turn on auto top-up.
• DELETE /v1/async/queue as a kill switch: it removes every task still QUEUED, which were never charged, and is safe to repeat.

Cost planning goes deeper; pricing lists the tiers.

Failure modes to test before production

• Webhook endpoint down for an hour: does the reconciler fill the gap?
• Submitter killed mid-batch and restarted: are there zero duplicates?
• Balance exhausted mid-cycle: does submission stop with an alert, and does coverage show the gap?
• Queue full: does the submitter back off using remainingCapacity?
• A prompt reworded in the catalog: does it get a new id and leave history intact?
• A brand alias added: does re-extraction from raw storage restate history?

Build order

1. Catalog, planner and plan table.
2. Submitter with idempotency keys.
3. Webhook receiver writing raw storage.
4. Reconciler.
5. Extractor and observation tables.
6. Views and dashboards, coverage first.

For a smaller first version, the batch pipeline tutorial is a single-script take on steps 2 to 5. To get an API key and run the first tasks, start with the quickstart and the async docs.

Keep reading