Open Geo

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    --brand "" --n-worker  \
          [--output dashboard|pdf|both] [--period today|all] [--lang en|ru|zh|ar]

Positional arguments

| arg | meaning | |---|---| | ` | Path to the input CSV. Columns: **query,lens** where lens ∈ general \| branded \| comparative. See examples/questions.csv for a ready sample. general = neutral query, no brand named; branded = brand explicitly named; comparative = brand vs alternatives. | | | Engine id, **snake_case**, e.g. google. This value is (a) the engine field written into every QueryCapture and the run, and (b) the basename of the capture playbook the workers load: engines/.md (so google ↔ engines/google.md). **This is the multi-engine extension point** — google (Google AI Overview), chatgptsearch (ChatGPT web search), claudesearch (Claude web search), yandexneuro (Yandex Alice / Нейро) and gemini (Google Gemini) ship today; the others (Perplexity, DeepSeek, …) are on the roadmap (ROADMAP Feature 3), and adding one is mainly authoring engines/.md (see engines/README.md). | | | The **target** domain to score. Accept any spelling (https://www.example.com, example.com); the pipeline normalizes it via pipeline.schema.normalizedomain`. Workers match the target with this same normalizer so matching is consistent. |

Flags

| flag | required | default | meaning | |---|---|---|---| | --brand "" | yes | — | Human brand name (free text, may contain spaces — keep it quoted). Stored on the run; used in report/dashboard titles and the summary. | | --n-worker | yes | — | Number of capture sub-agents to run in parallel — the run's concurrency. Step 2 splits the queries into N chunks, one per worker. | | --output dashboard\|pdf\|both | no | dashboard | Which deliverable(s) to produce in step 6. | | --period today\|all | no | all | Reporting window passed to the dashboard/report: today = just this run's date, all = full history for this brand+engine (enables the previous-run deltas described in INTERFACES §4.1). | | --lang en\|ru\|zh\|ar | no | en | UI language for the deliverables: it is passed to the report (report.generate --lang) and is the dashboard's default language (the switcher can still change it in the browser). Extensible to any code registered in i18n/locales.json. It also sets the language of the final summary you print in step 7. |

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 ` 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 ` (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, '', '')
print(json.dumps({'run_id': find_unfinished_run(conn, bid, '')}))
"

• 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 ` 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:

``bash .venv/bin/python -m pipeline.ingest \ --brand "" --domain --engine --new-run ``

stdout: {"run_id": } (per INTERFACES §3.1). Parse it and keep `` 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 ` (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/.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/.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): ``bash .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, )))) " ` 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/.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_.json), since parallel workers share /tmp.

• The target `**, the **--brand name**, and the **` 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/.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: ``bash .venv/bin/python -m pipeline.ingest --run-id .json ` Read stdout {"runid", "ok": [...], "skipped": [...], "errors": [...]} (INTERFACES §3.2). Ingest is **idempotent** on (runid, 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: ``bash .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=, n_queries=, n_ok=, n_failed=, status='done') " ` nqueries = total (query, lens) rows attempted (from the **full CSV**, including a resume's already-done rows); nok = rows captured (ingest keeps this live, = COUNT(results)); nfailed = nqueries − 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 

• 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 orchestr

…

Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

• Author: Pupok462
• Source: Pupok462/open-geo
• License: MIT

Install and usage instructions live in the source repository linked above.