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

arg	meaning
<questions.csv>	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. Either a hand-made CSV or one generated by STEP A.5 (question harvesting, Feature 1 — harvest/METHODOLOGY.md); both are first-class.
<engine>	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/<engine>.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 / Нейро), gemini (Google Gemini), deepseek (DeepSeek web search) and perplexity (Perplexity) ship today, all live-validated; the others are on the roadmap (ROADMAP Feature 3), and adding one is mainly authoring engines/<engine>.md (see engines/README.md).
<domain>	The target — a registrable domain (example.com) or a URL prefix (github.com/user/repo). Accept any spelling; normalized via pipeline.schema.normalize_target. Workers match links against the target via matches_target/target_ranks (same semantics pipeline-wide).

Flags

flag	required	default	meaning
--brand "<name>"	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 <N>	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 data|dashboard|pdf|both	no	data	Optional presentation output. A portable JSON artifact is always produced; data means no server and no PDF. dashboard, pdf, and both add those outputs.
--artifact-out <path.json>	no	reports/run-<run-id>.json	Absolute or caller-relative destination for the portable run artifact. Use this when another agent workflow needs the data in its own workspace.
--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 (adds the PDF trend chart / the dashboard's whole-period view). Previous-run deltas (INTERFACES §4.1) render whenever an earlier completed run exists — in the PDF for either period, and in the dashboard's latest-run view.
--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.
--force	no	off	Override the GEO-audit gate (STEP 0): proceed with the run even when the audit verdict is blocked (a category-A blocker — the domain is unreadable by the engine's search bot / unreachable / JS-only). Without it, a blocked verdict hard-stops before any run and prints the remediation. Advisory (ready_with_warnings) verdicts never need --force.
--repeat R	no	1	Repeat-run group (INTERFACES §2.1, Feature 5): capture the SAME question set R times as R ordinary runs sharing one group_id. Costs R× capture — a deliberate operator choice to separate signal from LLM noise. The dashboard then reads the group as one measurement: weighted mean of the seven metrics + a min–max spread chip per card (deltas are suppressed inside a group). R=1 = today's behavior, no group. See "Repeats" note under STEP 1.

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 ../../../.codex/agents/capture-worker.toml; 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>