2026-09-17, Engineering · Rank Tracking

Programmatic AI visibility tracking: the system design for tracking at scale

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

 prompt catalog ──► planner ──► plan table ──► submitter ──► POST /v1/async/task/batch
 (git, versioned)   (expand)    (expected      (≤500/batch)
                                 tasks)                          │
                                   ▲                             ▼
                                   │                     answer engines run
                                   │                             │
 dashboards ◄── metrics ◄── extractor ◄── raw store ◄── webhook receiver ◄──┘
                (SQL views)  (observations)  (JSON)       (verify, dedupe)
                                   ▲
                                   └──── reconciler (GET /v1/async/task/{id} for gaps)

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, COPILOT]
  markets: [{country: US}, {country: US, state: TX}, {country: GB}]
  runs_per_day: 2
  include: {searchQueries: true}
  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; the chat engines also accept 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-TX-2026-09-17-r1
  prompt_id       text not null,
  client          text not null,
  engine          text not null,           -- CHATGPT, GEMINI, COPILOT, ...
  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 credits (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

Three limits bound the system (rate limits):

• Concurrency slots per plan: how many tasks run at once. Async tasks wait in your queue for a slot; synchronous calls share the same slots and fail with CONCURRENT_LIMIT_EXCEEDED when all are busy. Don’t mix a large synchronous workload with a bulk run.
• Queue capacity: how many tasks may wait. GET /v1/async/status reports queuedTasks, processingTasks and a breakdown by priority.
• Task deadlines: a task has 72 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.

The throughput estimate follows from the first limit:

tasks per hour ≈ concurrency slots × 3600 / average task seconds

Take average task seconds from task.latencyMs on your own completed tasks, per engine; don’t assume a number. Then check the daily plan fits:

required slots ≈ tasks per day × average task seconds / (hours in your collection window × 3600)

If the plan doesn’t fit the window, 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 their slot risk the 72-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:

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 (3 credits per query for one page).
6. Accuracy: answers naming the client that contain a wrong key fact.

10. Cost control

Credits per async task: ChatGPT 5, plus 2 once if any of searchQueries, ads, shopping or rawResponse is included; Copilot 5; Grok 5 (7 with rawResponse); Gemini 4; Perplexity 4; AI Mode 4; Google Search 3 for the first page, 2 per extra page and 2 for AI Overview extraction. Synchronous calls add 2, so keep the pipeline fully async.

Monthly credits for one catalog line:

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

Worked example: 400 prompts, 2 markets, ChatGPT with search queries (7) and Copilot (5), 1 run per day, 30 days: 400 × 2 × 12 × 30 = 288,000 credits. Halving runs to every other day halves it; keeping include.searchQueries on one run a week and dropping it on the other 26 days saves 2 × 400 × 2 × 26 = 41,600. Reserve a separate budget for fan-out checks on Google.

Guards worth building:

• A pre-flight in the planner that sums expected credits per client and refuses a plan over budget.
• GET /v1/credits before each submission cycle; stop when remaining minus the next cycle’s plan goes negative.
• 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 plan sizes.

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?
• Credits 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