open-geo — GEO visibility run orchestrator

You are the orchestrator for one open-geo run: drive a list of queries through one AI engine, capture how the target domain shows up in the answers, ingest the captures through the validated pipeline, aggregate metrics, and emit a dashboard and/or a PDF report — finishing with a short summary.

This skill is the single operator entry point. It coordinates components that are specified in pipeline/INTERFACES.md (the authoritative contract). Read that file's §1 (capture contract) and §3 (CLI contracts) before acting if anything below is ambiguous — the shapes there win over this prose.

Conventions (from CLAUDE.md): code/identifiers and intermediate JSON are English. The final summary printed to the user follows `--lang` (default English). Work only inside the repository root (the directory of this repo / your current working directory). Run all Python with the project venv (.venv/bin/python) so pipeline.* imports resolve, with the repo root as the working directory (paths like data/aeo.db are repo-root-relative).

INVOCATION

/open-geo <questions.csv> <engine> <domain> --brand "<name>" --n-worker <N> \
          [--output dashboard|pdf|both] [--period today|all] [--lang en|ru|zh|ar]

Positional arguments

Flags

If a required argument is missing, go to STEP A (the parameter wizard) to collect it interactively. Only hard-stop — a short error (in --lang), no empty run — if a required value is still unresolved after the wizard (or the user abandons it), or if questions.csv does not exist / has no data rows.

STEP A — RESOLVE PARAMETERS (intro + wizard, with fast-path bypass)

Run this first, before STEP 0. Goal: end up with every required parameter resolved.

Required: questions.csv, engine, domain, --brand, --n-worker. Optional (defaults): --output (dashboard), --period (all), --lang (en).

1. Parse the invocation — gather values from positional args, flags, AND anything the user

expressed in free text (e.g. "measure example.com on google, 5 workers, pdf").

1. FAST PATH — all required resolved: do not print the intro or ask anything. Echo one

confirmation line — Running: csv=… engine=… domain=… brand=… n-worker=… output=… period=… lang=… — then proceed to STEP 0/1. (This is the path loops/headless use: pass full args, skip the wizard.)

1. GUIDED PATH — something required is missing:

a. Print a short intro (2–4 lines): what open-geo does (drives queries through an AI engine, measures the target domain's visibility/citation, emits a dashboard and/or PDF) and what it produces. b. Ask only for the missing parameters, using AskUserQuestion for the enumerable ones:

• engine — offer only engines that actually have a playbook:

.venv/bin/python -c "import glob,os; print('\n'.join(sorted(os.path.basename(p)[:-3] for p in glob.glob('engines/*.md') if os.path.basename(p)!='README.md')))" (today: google, chatgpt_search, claude_search, yandex_neuro, gemini). If the user names an engine without a playbook, say it is not available yet (ROADMAP Feature 3) and stop.

• --n-worker — presets 1 / 3 / 5 / 10 (+ custom).
• --output — dashboard / pdf / both. --period — today / all. --lang — en / ru / zh / ar.
• questions.csv — offer found CSVs (+ "other path"):

.venv/bin/python -c "import glob; print('\n'.join(glob.glob('*.csv')+glob.glob('examples/*.csv')))"

• domain and --brand — free text.

c. Echo the resolved parameters for a quick confirm, then proceed to STEP 0/1.

1. If a required value is still unknown after the wizard (or it is abandoned), apply the guard from

INVOCATION: a short error in --lang, no empty run.

STEP 0 — PRE-GATE HOOK (design-only placeholder — DO NOT IMPLEMENT)

No-op for v1. This section reserves the slot where the Domain GEO-Audit Gate (ROADMAP Feature 2) will run FIRST, before any run is created.

>

When that gate exists it will: audit <domain> for baseline GEO readiness (crawl access, robots.txt AI-bot policy, SSR vs JS-only content, sitemap, structured data, llms.txt, …), and — per ROADMAP — hard-stop only on category-A blockers (the site is physically unreadable by AI crawlers), emitting a remediation report of "what to add". Everything else is advisory (warn + continue, overridable with --force). The gate is expected to cache results per domain (TTL).

>

For now there is nothing to call. Do not implement the audit here and do not block the run. Treat this purely as the documented insertion point: if a future python -m audit.gate --domain <domain> (name TBD) is present and returns a hard blocker, this skill should abort before step 1 and surface the remediation report. Until then, proceed directly to step 1.

STEP 1 — CREATE OR RESUME THE RUN

First check for an unfinished run to resume — a previous run of this brand+engine left status='running' by a crash (INTERFACES §2.1). Look before creating anything:

.venv/bin/python -c "
import json
from pipeline.db import get_conn, init_db, get_or_create_brand, find_unfinished_run
conn = get_conn('data/aeo.db'); init_db(conn)
bid = get_or_create_brand(conn, '<name>', '<domain>')
print(json.dumps({'run_id': find_unfinished_run(conn, bid, '<engine>')}))
"

• `run_id` non-null → an unfinished run exists. Offer to resume it (reuse that

run_id; STEP 2 captures only the rows it is still missing) vs. start fresh. On the fast path (loops/headless, all args supplied) resume automatically — unattended recovery is the whole point. Keep the chosen <run_id> and skip the --new-run call.

• `run_id` null (or the user chose fresh) → create a fresh run and capture its

run_id from JSON stdout:

  .venv/bin/python -m pipeline.ingest \
    --brand "<name>" --domain <domain> --engine <engine> --new-run

stdout: {"run_id": <int>} (per INTERFACES §3.1). Parse it and keep <run_id> for every later step. Human/log noise goes to STDERR — only the JSON object is on STDOUT.

• If creation errors or stdout is not parseable JSON with a run_id, stop and report it

(in --lang). Nothing downstream can proceed without run_id.

STEP 2 — PREPARE THE WORK & THE PLAYBOOK

1. Read all data rows from <questions.csv> (header query,lens). Validate each lens

is one of general|branded|comparative; drop/flag malformed rows (note them for the summary). Let rows be the validated list, preserving file order.

1. Locate the capture playbook `engines/<engine>.md`. This file is the per-engine

capture instructions the subagents follow (e.g. engines/google.md for Google AI Overview — referenced in the house rules as "the capture playbook").

• If engines/<engine>.md is missing, do not invent a procedure. Stop and tell the

user (in --lang) that the playbook for this engine is not present yet and must be added before a run — the capture contract still applies, but the engine-specific "how to drive it" lives in that file. The pattern for authoring a new engine playbook is in engines/README.md (multi-engine is ROADMAP Feature 3). (`engines/google.md`, `engines/chatgpt_search.md`, `engines/claude_search.md`, `engines/yandex_neuro.md` and `engines/gemini.md` ship today; passing any other engine id needs its playbook written first.)

1. If resuming an existing run (STEP 1 returned one), drop rows already captured —

read the captured keys and keep only the missing (query, lens):

   .venv/bin/python -c "
   import json
   from pipeline.db import get_conn, get_captured_keys
   conn = get_conn('data/aeo.db')
   print(json.dumps(sorted(list(t) for t in get_captured_keys(conn, <run_id>))))
   "

Subtract those from rows. If nothing remains, skip capture entirely and jump to STEP 4.2 (finalize) → STEP 5. (Ingest is idempotent, so re-capturing a stored row is harmless — skipping just saves a browser hit.)

1. Split the rows to capture into min(N, len(rows)) contiguous chunks of roughly equal

size, where N = --n-worker. Each chunk keeps its rows' original (query, lens) pairs.

STEP 3 — FAN-OUT CAPTURE (one capture-worker subagent per chunk)

Spawn N = `--n-worker` subagents of type `capture-worker` (Task tool), one per chunk, and run them in parallel — each drives its chunk concurrently in its own browser tab/context. A capture worker's only job is to capture and RETURN data; it never ingests, creates runs, starts servers, or writes the DB. Its full step-by-step contract lives in .claude/agents/capture-worker.md — do not restate it here. Give each capture-worker a self-contained brief containing:

• The full text of engines/<engine>.md (the capture playbook).
• Its chunk of (query, lens) rows, and its chunk index (1..N) — used to name its

validation temp file uniquely (/tmp/open_geo_cap_<idx>.json), since parallel workers share /tmp.

• The target `<domain>`, the `--brand` name, and the `<engine>` id.
• A pointer to `pipeline/INTERFACES.md` §1 as the authoritative capture contract, and

to pipeline/schema.py :: QueryCapture / normalize_domain.

Do not give the worker the run_id, the DB path, or any ingest command — a capture worker never writes to the DB and never starts a server. The orchestrator owns all DB writes and the deliverables (steps 4 and 6).

The worker's full step-by-step contract — output fields, the no-DB and no-source-visit rules, per-worker temp-file self-validation, what to return — lives in .claude/agents/capture-worker.md. It is engine-agnostic; the injected engines/<engine>.md playbook is authoritative for how to drive the specific engine, and INTERFACES §1 for the QueryCapture shape.

Parallelism — N workers run concurrently

The skill spawns N = `--n-worker` capture sub-agents and runs them in parallel: step 2 splits the query rows into N chunks and each sub-agent drives its chunk concurrently, each in its own browser tab/context. --n-worker IS the run's real concurrency — raise it to go wider.

• If Google shows a reCAPTCHA / "unusual traffic" challenge, the affected worker

stops and surfaces it to the human (per the playbook) instead of solving or hammering it; the other workers keep going.

STEP 4 — INGEST & FINALIZE (orchestrator owns all DB writes)

The database is written only by you (the orchestrator), as each worker returns its chunk — incrementally, so a crash mid-run never loses already-captured work (INTERFACES §2.1). The workers never touched the DB.

1. Ingest each worker's chunk as it returns — incrementally, not one batch at the end

(durability: a crash can't lose chunks already returned). For each returned QueryCapture array, write it to a temp file (UTF-8/Cyrillic-safe) and ingest into the run:

   .venv/bin/python -m pipeline.ingest --run-id <run_id> < /tmp/open_geo_chunk_<idx>.json

Read stdout {"run_id", "ok": [...], "skipped": [...], "errors": [...]} (INTERFACES §3.2). Ingest is idempotent on (run_id, query, lens), so skipped (already-stored rows — normal on a resume/retry) is safe, never a duplicate. Fix any row in errors — correct the field from the returned data, or re-dispatch that one (query, lens) to a worker — and re-send only the fixed objects to the same --run-id. Repeat until errors is empty (bounded retries; then report residual failures).

1. Finalize counts + status. There is no "finalize" CLI; use the documented helper

pipeline.db.update_run_counts (INTERFACES §2) inline:

   .venv/bin/python -c "
   from pipeline.db import get_conn, update_run_counts
   conn = get_conn('data/aeo.db')
   update_run_counts(conn, run_id=<run_id>,
                     n_queries=<total rows attempted>,
                     n_ok=<rows accepted by ingest>,
                     n_failed=<rows never accepted>,
                     status='done')
   "

n_queries = total (query, lens) rows attempted (from the full CSV, including a resume's already-done rows); n_ok = rows captured (ingest keeps this live, = COUNT(results)); n_failed = n_queries − n_ok. Set status='done' on success, or 'failed' if the run collapsed (playbook missing, engine unreachable for everything). Finalizing `status` is the orchestrator's job — `ingest` never sets it (INTERFACES §2.1/§3.2); a completed run with status='done' unlocks previous-run deltas for --period all (INTERFACES §4.1). Never leave a run stuck in `status='running'`.

STEP 5 — AGGREGATE METRICS

.venv/bin/python -m pipeline.aggregate --run-id <run_id>

• Computes metrics per lens plus one lens="all" aggregate row, writes them to the

metrics table, and prints a JSON summary on stdout (INTERFACES §3.3). Capture this stdout — step 7's summary reads its metrics (lens="all" row) directly.

• In the same pass it also builds the top-domains leaderboard into domain_stats

(INTERFACES §2/§4.2): for every domain in sources/citations (not just the target) — appearances + average source/citation position, per lens + all. This is deterministic math (no extra step for you); the summary's top_domains echoes the all-scope top 10. It powers the dashboard's "Top domains in answer space" panel and the report's top-domains section, and recomputes idempotently on re-aggregate.

STEP 5b — SYNTHESIZE PER-LENS SENTIMENT (orchestrator writes the qualitative roll-up)

pipeline.aggregate (STEP 5) stays deterministic math — it does not touch sentiment. You (the orchestrator, already an LLM) write the qualitative per-lens roll-up here, then persist it via pipeline.lens_sentiment (INTERFACES §3.4) into the lens_sentiment table (INTERFACES §2). This is separate from metrics on purpose, so a re-aggregate never clobbers the synthesized prose.

1. Gather the per-query `sentiment`s grouped by lens for this run. You already have them

from the STEP 4 captures; if not handy, read them back from results inline:

   .venv/bin/python -c "
   import json
   from pipeline.db import get_conn
   conn = get_conn('data/aeo.db')
   rows = conn.execute(
       'SELECT lens, sentiment FROM results WHERE run_id=? ORDER BY lens',
       (<run_id>,)).fetchall()
   print(json.dumps([dict(r) for r in rows], ensure_ascii=False))
   "

1. Write ONE short, neutral sentence per lens that appears in the run (general,

branded, comparative), plus an all synthesis across them. Summarize ONLY what the per-query `sentiment` strings of that lens actually say — never invent ranks, competitors, numbers, or praise the captures don't contain; keep it ~1 sentence.

• Language: follow the DATA, not `--lang`. The summary is a roll-up of captured sentiment

text, so write it in the language those sentiment strings are in (e.g. Russian captures → Russian summary), regardless of the deliverable --lang.

• If a lens had the brand in no query (every sentiment null), set that lens's summary

to `null` (the UI then shows a "not mentioned" fallback). Likewise all is null only if the brand appeared in no query at all.

1. Persist by piping a JSON object {lens: summary} to pipeline.lens_sentiment.

Write the JSON to a temp file first for UTF-8/Cyrillic safety, exactly like the STEP 4 batch ingest does:

   # /tmp/open_geo_sentiment.json holds e.g.
   # {"all": "...", "general": "...", "branded": "...", "comparative": null}
   .venv/bin/python -m pipeline.lens_sentiment --run-id <run_id> < /tmp/open_geo_sentiment.json

Read stdout {"run_id": <run_id>, "written": [...]} (INTERFACES §3.4) to confirm which lenses were upserted. Only the lenses you include are written; an unknown run_id exits 1.

The dashboard then renders these as a "Sentiment by lens" card strip above the results table, and the PDF report shows them as the lead line of its sentiment section.

STEP 6 — EMIT DELIVERABLE(S) per --output

Ordering — the skill does this, not a worker, and only after steps 3–5. Deliverables are produced by the orchestrator once every capture is collected & ingested, the run is finalized, and metrics are aggregated. A capture worker never starts a server or generates a report. Start long-running servers in the background on a free port.

>

The report and dashboard components are built and their entry points are verified working (commands below are the real ones). They are intentionally not in INTERFACES — their contracts live in their own dirs (report/generate.py and dashboard/README.md). If you need detail beyond what's shown, read those.

--output dashboard (default) — or as part of both

Start the dashboard (FastAPI backend + Vite/React frontend) and print the local URL. The frontend selects brand/engine/period through its own UI controls (read from the API), so you do not scope brand/engine/period via the query string — only the UI language: hand the operator http://localhost:5173/?lang=<lang>, which seeds the dashboard's initial language from the run's --lang (the in-browser switcher still overrides it, and the choice persists in localStorage).

# Run BOTH in the background (they are long-running dev servers).
# Background shells do NOT inherit the repo-root CWD, so use ABSOLUTE paths anchored at
# <REPO> = the repository root (your working directory). Do NOT use a relative
# `.venv/bin/python` or `cd dashboard/web` here — backgrounded, they fail (exit 127 /
# wrong CWD). `--app-dir <REPO>` lets uvicorn import `dashboard.api` regardless of CWD.
#
# 1) API (read-only over data/aeo.db). PICK A FREE PORT — 8000 is often taken:
OPEN_GEO_DB=<REPO>/data/aeo.db <REPO>/.venv/bin/python -m uvicorn dashboard.api:app \
    --host 127.0.0.1 --port <PORT> --app-dir <REPO>

# 2) Web (Vite dev server), pointed at the API's port. Use `npm --prefix` instead of `cd`
#    (run `npm --prefix <REPO>/dashboard/web install` once if node_modules is missing):
VITE_API_BASE=http://127.0.0.1:<PORT> npm --prefix <REPO>/dashboard/web run dev

• Port caveat: local port 8000 is often already occupied by another service on

this machine. Pick a free port for the API (e.g. 8077) and point the frontend at it via VITE_API_BASE (CORS is open, so a cross-origin base works without the dev proxy):

  VITE_API_BASE=http://127.0.0.1:<PORT> npm --prefix <REPO>/dashboard/web run dev

• Verify before handing off (a backgrounded server can exit non-zero or the port can

clash): probe both before printing the URL —

  curl -s http://127.0.0.1:<PORT>/api/health
  curl -s -o /dev/null -w '%{http_code}\n' http://localhost:5173/

so you surface a working URL, not a hopeful one.

• After both are up, print the Vite dev URL the operator should open —

http://localhost:5173/?lang=<lang> (the frontend's own controls drive brand/engine/period; ?lang=<lang> seeds the UI language from the run's --lang, and the switcher still overrides). If dashboard/ cannot be started, say so (in --lang) and skip gracefully (still finish steps 5 and 7).

--output pdf — or as part of both

.venv/bin/python -m report.generate \
  --brand "<name>" --domain <domain> --engine <engine> \
  --period <period> --lang <lang> \
  --out reports/<brand>_<date>.pdf [--db data/aeo.db]

• This is the real, built CLI. It prints progress/status to stderr; the output path

(--out) is what to surface to the operator. Pass --lang <lang> (the run's --lang, default en) so the report renders in that language.

• Use <date> = today (YYYY-MM-DD). Create reports/ if missing. **Print the resulting

file path.** If the command fails, say so (in --lang) and skip gracefully.

--output both

Do both of the above: start the dashboard (print URL) and generate the PDF (print path).

STEP 7 — SUMMARY (printed to the user, in --lang)

Read the `lens="all"` row from the pipeline.aggregate JSON captured in step 5 and print a short summary of headline metrics for this run, in the `--lang` language (default English). Cover:

• AI Overview coverage (overview_coverage) — share of queries where an AI Overview

rendered at all.

• Visibility in sources (visibility_in_sources) — share of overview queries where the

target domain made it into sources (n_in_sources / n_overviews).

• Visibility in citations (visibility_in_citations) — share of overview queries where

the domain is cited in the answer (n_cited / n_overviews).

• Average source position (avg_source_position) — average best (min) rank of the

domain among sources (lower = better; — if the domain never appears in sources).

• Average citation position (avg_citation_position) — average best (min) rank of the

domain among citations (lower = better; — if the domain is never cited).

• Relative citation (relative_citation) — the source→citation conversion: of the

queries where the domain was in sources, the share where it was actually cited (n_cited / n_in_sources; higher = better, ∈ [0, 1]; — if the domain never appears in sources). This is the last step of the visibility funnel (n_cited ≤ n_in_sources ≤ n_overviews ≤ n_queries).

Format as percentages where natural, and note guard cases (null → "no data" / "—", not 0). End by pointing to the produced deliverable(s): the dashboard URL and/or the PDF path. If --period all and a previous completed run exists, you may mention the direction of change (deltas are computed at read-time per INTERFACES §4.1) — otherwise omit.

Example shape (English; fill with real numbers; one lens="all" row drives it):

Run for brand "Example" (engine google), queries: 30.
• AI Overview coverage: 73% (22 of 30 queries).
• Visibility in sources: 41% of overview queries.
• Visibility in citations: 32% of overview queries.
• Average source position: 2.4 (lower is better).
• Average citation position: 1.7 (lower is better).
• Source→citation conversion (relative citation): 78% (higher is better).
Report: reports/example_2026-06-19.pdf · Dashboard: http://localhost:5173/?lang=en

COMPONENT DEPENDENCY MAP (where this skill leans on others)

Keep the run operator-friendly: parse JSON from stdout (never scrape logs), fail loudly (in --lang) on missing prerequisites, and never leave a run stuck in status='running'.