# Open Geo

> Run a list of queries through a chosen AI engine, measure the target domain's visibility/citation in the AI answers, and produce a dashboard or PDF report. Use when the user runs /open-geo or asks to measure a brand's GEO / AI-search visibility (citations in Google AI Overview, etc.).

- **Type:** Skill
- **Install:** `agentstack add skill-pupok462-open-geo-open-geo`
- **Verified:** Yes — security-reviewed for prompt injection and unsafe behavior
- **Seller:** [Pupok462](https://agentstack.voostack.com/s/pupok462)
- **Installs:** 0
- **Category:** [Data & Analytics](https://agentstack.voostack.com/c/data-and-analytics)
- **Latest version:** 0.1.0
- **License:** MIT
- **Upstream author:** [Pupok462](https://github.com/Pupok462)
- **Source:** https://github.com/Pupok462/open-geo/tree/main/.claude/skills/open-geo

## Install

```sh
agentstack add skill-pupok462-open-geo-open-geo
```

Requires the [AgentStack CLI](https://agentstack.voostack.com/docs/cli). Works with Claude Code, Cursor, and any MCP-compatible agent.

## About

# 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), `chatgpt_search` (ChatGPT web search), `claude_search` (Claude web search), `yandex_neuro` (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.normalize_domain`. 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").
2. **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.)
3. **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.
4. 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:

```bash
.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.
2. 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.)*
3. **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.)
4. 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 `{"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).

2. **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')
   "
   ```
   `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

```bash
.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](https://github.com/Pupok462)
- **Source:** [Pupok462/open-geo](https://github.com/Pupok462/open-geo)
- **License:** MIT

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

## Pricing

- **Free** — Free

## Security capabilities

Automated source analysis of v0.1.0 — what this tool can access:

- **Network access:** yes
- **Filesystem access:** no
- **Shell / process execution:** no
- **Environment & secrets:** no
- **Dynamic code execution:** no

*"Yes" means the capability is present in the source — more access means more to trust, not that it is unsafe.*


## Versions

- **0.1.0** — security scan: passed — Imported from the upstream source.

## Links

- Listing page: https://agentstack.voostack.com/l/skill-pupok462-open-geo-open-geo
- Seller: https://agentstack.voostack.com/s/pupok462
- Browse the marketplace: https://agentstack.voostack.com/browse

---
Listed on AgentStack — the marketplace for AI agent skills and MCP servers. Every listing is security-reviewed. Creators keep 70%.