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 portable JSON artifact plus any requested presentation output — finishing with a short summary.

This skill is the single operator and agent-workflow entry point. It can be invoked directly by a user or called as one step inside another agent's workflow; in both cases it returns the same versioned JSON artifact for downstream consumption. 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.

Code/identifiers and intermediate JSON are English. The final summary printed to the user follows --lang (default English). Run pipeline commands from the resolved open-geo runtime root with its project venv (.venv/bin/python) so pipeline.* imports resolve. An explicit absolute --artifact-out may point into the caller's workspace; all other runtime state stays inside open-geo.

INVOCATION

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

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 R — RESOLVE & BOOTSTRAP THE RUNTIME (always first)

The user should not have to clone the repository, run setup, start Python, or launch a dashboard manually. Resolve one runtime root and do the reversible setup yourself:

1. Prefer the current working directory when it contains pipeline/INTERFACES.md.
2. Otherwise prefer a valid OPEN_GEO_ROOT supplied by the caller.
3. Otherwise use the installed plugin/package root when the host exposes it, it contains pipeline/INTERFACES.md, and it is writable (for Claude Code this is ${CLAUDE_PLUGIN_ROOT}). A read-only package root falls through to the managed runtime.
4. Otherwise use ${OPEN_GEO_HOME:-$HOME/.local/share/open-geo}/runtime. If it does not exist, create its parent and clone https://github.com/Pupok462/open-geo there. This is an implementation detail of the skill, not a manual prerequisite for the user.
5. If the chosen root has no executable .venv/bin/python, run scripts/setup.sh --minimal from that root. For --output dashboard or both, run the full scripts/setup.sh if dashboard/web/node_modules is absent. Never install dashboard dependencies for the default data mode.

Before changing directories, remember the caller's original working directory. Resolve a relative --artifact-out against that original directory, not the runtime root. After this step, change the command working directory to the runtime root and use absolute paths when reporting artifacts. If bootstrap fails (no Git/Python/network or dependency error), stop with the exact failed command and remediation; do not create an empty run. A logged-in browser session may still require the user to authenticate once, but they never need to launch open-geo services themselves.

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

Run this after the STEP R guard, before STEP 0. Goal: end up with every required parameter resolved.

Required: questions.csv, engine, domain, --brand, --n-worker. Optional (defaults): --output (data), --artifact-out (reports/run-<run-id>.json), --period (all), --lang (en), --force (off — overrides a blocked audit-gate verdict, STEP 0), --repeat (1 — R independent captures of the same CSV under one group tag, STEP 1).

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, sorted: chatgpt_search, claude_search, deepseek, gemini, google, perplexity, yandex_neuro). 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 — data / dashboard / pdf / both. --period — today / all. --lang — en / ru / zh / ar.
   • questions.csv — offer found CSVs (+ "other path"), and a "Generate a set" option: .venv/bin/python -c "import glob; print('\n'.join(glob.glob('*.csv')+glob.glob('examples/*.csv')))" If the user picks Generate, leave questions.csv unresolved here and let STEP A.5 harvest it (it writes the CSV and sets the path). If they pick a file / give a path, that is the input CSV and STEP A.5 is skipped.
   • 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 — GEO-AUDIT GATE (runs FIRST: after the domain is known, before harvesting or a run)

Run this right after STEP A (so <domain> and <engine> are resolved) and before STEP A.5 and STEP 1 — there is no point harvesting questions or spending capture tokens on a domain an AI engine cannot even read. This is the Domain GEO-Audit Gate (ROADMAP Feature 2); the contract is pipeline/INTERFACES.md §7, the check semantics audit/CHECKS.md. It is deterministic Python (non-LLM, no browser).

1. Run the audit — it fetches robots.txt / homepage / sitemap.xml / llms.txt / /.well-known, grades each check by severity, and writes the result to the audits table so the PDF/dashboard can show it later:

   .venv/bin/python -m audit.gate --domain <domain> --engine <engine>
   

   Parse stdout — a single AuditResult JSON (INTERFACES §7.1): verdict (ready | ready_with_warnings | blocked), score (0–100), passed, blockers (check ids), and checks[] (each id, severity, status, detail, remediation). A human summary is on STDERR. Add --no-cache to force a fresh audit (by default a recent audit for the same domain is reused within its TTL).

2. Decide, per verdict:

   • blocked (a category-A blocker failed — the site is unreachable, non-200, JS-only, or robots.txt blocks the engine's search bot) and no --force given: hard-stop before any run. Print (in --lang) a short remediation report — for each blocker its detail + the concrete remediation fix, then the advisory warn/fail checks below it — and say plainly: the domain is not visibility-ready, so a capture run would waste tokens; fix the blockers, or re-run with --force to measure anyway. Do not create a run and do not harvest. Stop.
   • blocked with --force: warn loudly (list the blockers + their fixes), then continue — the operator chose to measure an unready domain.
   • ready_with_warnings: briefly surface the advisory problems (the warn/fail checks with their detail) and the score, then continue to STEP A.5.
   • ready: one line — GEO-audit: ready (score N/100) — continue.
   • The gate itself failed (exit code 1, no JSON on stdout — the domain string is unusable, or nothing could be fetched at all): this is "unknown", not "blocked", and an unknown premise never blocks (same rule as the skip statuses in audit/CHECKS.md). Print the gate's STDERR line, say plainly that domain readiness could not be verified, and continue to STEP A.5 — but if the failure looks like a typo in <domain> (unresolvable host, stray characters), confirm the target with the user first rather than measuring the wrong domain.

3. The audit is now stored (keyed by the registrable domain), so STEP 6's PDF/dashboard read it back (get_latest_audit) and render the full check table — you need not repeat the audit there.

Boundary. The gate is deterministic and only emits structured JSON; you (the orchestrator) turn that JSON into the human-language remediation the operator reads — the same division as the lens_sentiment prose vs the aggregate math. Only category-A failures block; everything else is advisory. Authority: pipeline/INTERFACES.md §7 + audit/CHECKS.md.

STEP A.5 — SOURCE THE QUESTIONS (bring-your-own vs harvest a grounded set)

Run this after STEP A and STEP 0, before STEP 1. Goal: end up with a real <questions.csv> on disk.

1. FAST PATH / bring-your-own — a real CSV is already resolved. If STEP A resolved <questions.csv> to a path that exists and has data rows, this step is a no-op — use that file and go straight to STEP 1. (A user's own hand-made query,lens CSV is a first-class input; loops/headless always take this path.)

   Hand-off from a core build. If you were handed a core.json instead (INTERFACES §8 — written by demand.core, typically by the semantic-core skill), read questions_csv, brand and domain out of it and take this same fast path. The CSV it points at is an ordinary query,lens file; nothing downstream distinguishes it. Mention the core's totals.coverage in the run summary so the operator knows how much of the set rests on measured volume.

2. GENERATE PATH — the user chose "Generate a set" (or no CSV is resolved). Harvest one: read references/harvest.md now and follow it — it carries the full procedure (segment planning, the harvest-worker fan-out, the demand gate, the skeptic pass, harvest.build, the rationale file, and the human review gate). Harvesting is agentic and opt-in; the process authority is harvest/METHODOLOGY.md, the contract is pipeline/INTERFACES.md §6.

Boundary. Harvesting only produces the CSV; nothing downstream changes. The capture contract (§1), the run, ingest/aggregate are untouched — STEP 1 onward treats a harvested CSV exactly like a hand-made one.

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 -m pipeline.run --resume-check \
  --brand "<name>" --domain <domain> --engine <engine> --csv <questions.csv>

stdout: {"run_id", "resumable", "run_at", "n_captured", "n_missing"} (INTERFACES §3.7).

• run_id non-null and resumable true → the unfinished run holds a subset of THIS question set. 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 non-null but resumable false → do NOT resume, create a fresh run. The unfinished run was captured from a different question set; appending this CSV to it would blend two question sets under one run_id and score them as one measurement. Say plainly which run was left behind (run_id, run_at, n_captured) so the user can finish or drop it later, then continue as if run_id were null.

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

Repeats (--repeat R, R > 1) — R independent captures of the same CSV under one group_id, so readers see mean + spread instead of one noisy run (INTERFACES §2.1). Read references/deliverables.md for the flow; R=1 (the default) needs nothing extra.

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.
2. 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, engines/gemini.md, engines/deepseek.md and engines/perplexity.md ship today; passing any other engine id needs its playbook written first.)
3. If resuming an existing run (STEP 1 returned one), capture only what is still missing — the pending rows come back in file order:

   .venv/bin/python -m pipeline.run --pending --run-id <run_id> --csv <questions.csv>
   

   stdout: {"run_id", "n_total", "n_captured", "n_pending", "pending": [[query, lens], …]} (INTERFACES §3.7). Use pending as 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 (Agent tool) — one per chunk, all in one message so they run concurrently, each driving its chunk in its own browser tab/context. --n-worker IS the run's real concurrency; raise it to go wider.

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 — output fields, the no-DB and no-source-visit rules, per-worker temp-file self-validation, what to return — lives in .agentsmesh/agents/capture-worker.md; do not restate it. Give each worker a self-contained brief containing:

• The full text of engines/<engine>.md (the capture playbook — authoritative for how to drive this specific engine).
• 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).

• If the engine 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).

2. Finalize counts + status (INTERFACES §3.7):

   .venv/bin/python -m pipeline.run --finalize --run-id <run_id> \
     --n-queries <total rows attempted> --n-ok <rows accepted by ingest> --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 defaults to the difference. Use --status 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); only runs with status='done' feed previous-run deltas and the --period all rollup (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 sentiments grouped by lens for this run. You already have them from the STEP 4 captures; if not handy, read them back (INTERFACES §3.7):

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

2. 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.
3. 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 — EXPORT DATA, THEN ADD OPTIONAL PRESENTATION OUTPUTS

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 exports the run, starts a server, or generates a report.

Always — portable JSON run artifact

.venv/bin/python -m pipeline.artifact \
  --run-id <run_id> --db data/aeo.db \
  --out <artifact-out-or-reports/run-<run_id>.json>

Parse stdout as JSON and retain artifact_path. The artifact schema is open-geo.run-artifact.v1 and contains run metadata, brand/target, metrics by lens, qualitative lens summaries, decoded per-query captures, per-lens domain statistics, and the latest matching audit. This file is the handoff contract for other agents: downstream steps consume it instead of scraping the human summary, querying SQLite directly, or keeping the dashboard running.

For the default --output data, stop presentation work here and continue to STEP 7. No FastAPI/Vite process is started and no browser window needs to remain open after capture.

dashboard · pdf · both — or --repeat R > 1

Read references/deliverables.md and follow it. It carries the verified commands and their caveats: the dashboard's two background servers (absolute paths, a free port, the curl health probe before you hand over a URL), report.generate including the combined --engines all document, and the per-repeat artifact naming. These presentation contracts intentionally live in their own dirs (report/generate.py, dashboard/README.md) rather than in INTERFACES. If a deliverable cannot be produced, say so (in --lang) and skip gracefully — still finish steps 5 and 7.

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): answer coverage (overview_coverage), visibility in sources (visibility_in_sources), visibility in citations (visibility_in_citations), average source / citation position (lower = better), relative citation (relative_citation — the source→citation conversion, higher = better) and brand mention rate (brand_mention_rate — an adjacent axis, not a funnel stage). For the precise reading of any of them, see references/metrics.md (authority: INTERFACES §4).

Format as percentages where natural, and note guard cases (null → "no data" / "—", not 0). End by pointing to the absolute JSON artifact path, then the dashboard URL and/or PDF path when requested. If 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.
• Answer coverage: 73% (22 of 30 queries).
• Visibility in sources: 41% of grounded answers.
• Visibility in citations: 32% of grounded answers.
• 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).
• Brand mention rate: 55% of grounded answers name the brand.
Data: /absolute/path/reports/run-42.json
Report: /absolute/path/reports/example_2026-08-18.pdf · Dashboard: http://localhost:5173/?lang=en

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'.